Send advertised host and port in open (#40)

* Send Send advertised host and port in open

https://github.com/rabbitmq/rabbitmq-server/pull/3060/files
fix offest client side and test

* Change error handling

remove pre-declared option on the perf test, in the same way as Java client does. It checks the PreconditionFailed
condition. cc @gerhard

* Change test

* Change socket return type

Author: Gsantomaggio

Makefile

@@ -23,7 +23,7 @@ check: $(STATICCHECK)
 	$(STATICCHECK) ./pkg/stream
 
 test: vet fmt check
-	go test -v  ./pkg/stream -race -coverprofile=coverage.txt -covermode=atomic
+	go test -v  ./pkg/stream -race -coverprofile=coverage.txt -covermode=atomic -tags debug
 
 integration-test: vet fmt check
 	go test -v  ./pkg/system_integration -race -coverprofile=coverage.txt -covermode=atomic -tags debug

examples/getting_started.go

@@ -53,7 +53,6 @@ func main() {
 	)
 
 	CheckErr(err)
-
 	//optional publish confirmation channel
 	chPublishConfirm := make(chan []int64, 1)
 	go func(ch chan []int64) {
@@ -81,10 +80,10 @@ func main() {
 	//}, nil)
 	// if you need to track the offset you need a consumer name like:
 	handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
-		fmt.Printf("consumer id: %d, text: %s \n ", consumerContext.Consumer.ID, message.Data)
+		fmt.Printf("consumer name: %s, text: %s \n ", consumerContext.Consumer.GetConsumerName(), message.Data)
 		err := consumerContext.Consumer.Commit()
 		if err != nil {
-			fmt.Printf("Error during commit")
+			fmt.Printf("Error during commit: %s", err)
 		}
 	}
 
@@ -96,7 +95,8 @@ func main() {
 		fmt.Printf("Consumer: %s closed on the stream: %s, reason: %s \n", event.Name, event.StreamName, event.Reason)
 	}()
 
-	consumer, err := env.NewConsumer(context.TODO(), streamName,
+	consumer, err := env.NewConsumer(context.TODO(),
+		streamName,
 		handleMessages,
 		channelClose,
 		stream.NewConsumerOptions().
@@ -106,7 +106,7 @@ func main() {
 
 	fmt.Println("Press any key to stop ")
 	_, _ = reader.ReadString('\n')
-	err = consumer.UnSubscribe()
+	err = consumer.Close()
 	time.Sleep(200 * time.Millisecond)
 	CheckErr(err)
 	err = env.Close()

examples/offset/offset.go

@@ -0,0 +1,90 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"github.com/google/uuid"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
+	"os"
+	"time"
+)
+
+func CheckErr(err error) {
+	if err != nil {
+		fmt.Printf("%s ", err)
+		os.Exit(1)
+	}
+}
+
+func CreateArrayMessagesForTesting(batchMessages int) []*amqp.Message {
+	var arr []*amqp.Message
+	for z := 0; z < batchMessages; z++ {
+		arr = append(arr, amqp.NewMessage([]byte("1234567890")))
+	}
+	return arr
+}
+
+func main() {
+	reader := bufio.NewReader(os.Stdin)
+	// Set log level, not mandatory by default is INFO
+	//stream.SetLevelInfo(stream.DEBUG)
+
+	fmt.Println("Getting started with Streaming client for RabbitMQ")
+	fmt.Println("Connecting to RabbitMQ streaming ...")
+
+	env, err := stream.NewEnvironment(
+		stream.NewEnvironmentOptions().
+			SetHost("localhost").
+			SetPort(5552).
+			SetUser("guest").
+			SetPassword("guest").
+			SetMaxConsumersPerClient(1))
+	CheckErr(err)
+	// Create a stream, you can create streams without any option like:
+	// err = env.DeclareStream(streamName, nil)
+	// it is a best practise to define a size,  1GB for example:
+	streamName := uuid.New().String()
+	err = env.DeclareStream(streamName,
+		&stream.StreamOptions{
+			MaxLengthBytes: stream.ByteCapacity{}.GB(2),
+		},
+	)
+
+	CheckErr(err)
+
+	producer, err := env.NewProducer(streamName, nil, nil)
+	CheckErr(err)
+
+	go func() {
+		for i := 0; i < 2; i++ {
+			_, err = producer.BatchPublish(context.Background(), CreateArrayMessagesForTesting(100))
+			time.Sleep(1 * time.Second)
+		}
+	}()
+
+	counter := 0
+	handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
+		counter = counter + 1
+		fmt.Printf("messages consumed: %d \n ", counter)
+	}
+
+	consumer, err := env.NewConsumer(context.TODO(), streamName,
+		handleMessages,
+		nil,
+		stream.NewConsumerOptions().
+			SetConsumerName("my_consumer").                      // set a consumer name
+			SetOffset(stream.OffsetSpecification{}.Offset(100))) // start specific offset, in this case we start from the 100 so it will consume 100 messages
+	CheckErr(err)
+
+	fmt.Println("Press any key to stop ")
+	_, _ = reader.ReadString('\n')
+	err = producer.Close()
+	CheckErr(err)
+	err = consumer.Close()
+	CheckErr(err)
+	err = env.DeleteStream(streamName)
+	CheckErr(err)
+
+}

examples/publishersError/publisherError.go

@@ -39,7 +39,7 @@ func main() {
 
 		for {
 			pError := <-ch
-			fmt.Printf("Error during publish message id:%d,  error: %s \n", pError.PublishingId, pError.ErrorMessage)
+			fmt.Printf("Error during publish message id:%d,  error: %s \n", pError.PublishingId, pError.Err)
 		}
 
 	}(chPublishError)

perfTest/cmd/commands.go

@@ -28,7 +28,6 @@ var (
 	consumers           int
 	publishersPerClient int
 	consumersPerClient  int
-	preDeclared         bool
 	streams             []string
 	maxLengthBytes      string
 	maxSegmentSizeBytes string
@@ -54,7 +53,6 @@ func setupCli(baseCmd *cobra.Command) {
 	baseCmd.PersistentFlags().IntVarP(&rate, "rate", "", 0, "Limit publish rate")
 	baseCmd.PersistentFlags().IntVarP(&variableRate, "variable-rate", "", 0, "Variable rate to value")
 	baseCmd.PersistentFlags().IntVarP(&variableBody, "variable-body", "", 0, "Variable body size")
-	baseCmd.PersistentFlags().BoolVarP(&preDeclared, "pre-declared", "", false, "Pre created stream")
 	baseCmd.PersistentFlags().BoolVarP(&exitOnError, "exit-on-error", "", true, "Close the app in case of error")
 	baseCmd.PersistentFlags().BoolVarP(&printStatsV, "print-stats", "", true, "Print stats")
 	baseCmd.PersistentFlags().StringSliceVarP(&streams, "streams", "", []string{"perf-test-go"}, "Stream names")

perfTest/cmd/silent.go

@@ -138,33 +138,31 @@ func randomSleep() {
 }
 
 func initStreams() error {
-	if !preDeclared {
-		logInfo("Declaring streams: %s", streams)
-		env, err := stream.NewEnvironment(stream.NewEnvironmentOptions().SetUri(
-			rabbitmqBrokerUrl))
-		if err != nil {
-			logError("Error init stream connection: %s", err)
-			return err
-		}
+	logInfo("Declaring streams: %s", streams)
+	env, err := stream.NewEnvironment(stream.NewEnvironmentOptions().SetUri(
+		rabbitmqBrokerUrl))
+	if err != nil {
+		logError("Error init stream connection: %s", err)
+		return err
+	}
 
-		for _, streamName := range streams {
+	for _, streamName := range streams {
 
-			err = env.DeclareStream(
-				streamName,
-				stream.NewStreamOptions().
-					SetMaxLengthBytes(stream.ByteCapacity{}.From(maxLengthBytes)).
-					SetMaxSegmentSizeBytes(stream.ByteCapacity{}.From(maxSegmentSizeBytes)))
-			if err != nil {
-				logError("Error declaring stream: %s", err)
+		err = env.DeclareStream(
+			streamName,
+			stream.NewStreamOptions().
+				SetMaxLengthBytes(stream.ByteCapacity{}.From(maxLengthBytes)).
+				SetMaxSegmentSizeBytes(stream.ByteCapacity{}.From(maxSegmentSizeBytes)))
+		if err != nil {
+			if err == stream.PreconditionFailed {
+				logError("The stream: %s already exist with different parameters", streamName)
 				_ = env.Close()
 				return err
 			}
 		}
-		logInfo("End Init streams :%s\n", streams)
-		return env.Close()
 	}
-	logInfo("Predeclared streams: %s\n", streams)
-	return nil
+	logInfo("End Init streams :%s\n", streams)
+	return env.Close()
 }
 func startPublishers() error {
 	env, err := stream.NewEnvironment(stream.NewEnvironmentOptions().SetUri(

pkg/stream/client.go

@@ -139,7 +139,7 @@ func (c *Client) peerProperties() error {
 		writeString(b, element)
 	}
 
-	return c.handleWrite(b.Bytes(), resp)
+	return c.handleWrite(b.Bytes(), resp).Err
 }
 
 func (c *Client) authenticate(user string, password string) error {
@@ -193,14 +193,14 @@ func (c *Client) sendSaslAuthenticate(saslMechanism string, challengeResponse []
 	writeInt(b, len(challengeResponse))
 	b.Write(challengeResponse)
 	err := c.handleWrite(b.Bytes(), resp)
-	if err != nil {
-		return err
+	if err.Err != nil {
+		return err.Err
 	}
 	// double read for TUNE
 	tuneData := <-respTune.data
-	err = c.coordinator.RemoveResponseByName("tune")
-	if err != nil {
-		return err
+	errR := c.coordinator.RemoveResponseByName("tune")
+	if errR != nil {
+		return errR
 	}
 
 	return c.socket.writeAndFlush(tuneData.([]byte))
@@ -214,7 +214,18 @@ func (c *Client) open(virtualHost string) error {
 	writeProtocolHeader(b, length, commandOpen,
 		correlationId)
 	writeString(b, virtualHost)
-	return c.handleWrite(b.Bytes(), resp)
+	err := c.handleWriteWithResponse(b.Bytes(), resp, false)
+	if err.Err != nil {
+		return err.Err
+	}
+
+	advHostPort := <-resp.data
+	for k, v := range advHostPort.(ClientProperties).items {
+		c.clientProperties.items[k] = v
+	}
+	_ = c.coordinator.RemoveResponseById(resp.correlationid)
+	return nil
+
 }
 
 func (c *Client) DeleteStream(streamName string) error {
@@ -227,7 +238,7 @@ func (c *Client) DeleteStream(streamName string) error {
 
 	writeString(b, streamName)
 
-	return c.handleWrite(b.Bytes(), resp)
+	return c.handleWrite(b.Bytes(), resp).Err
 }
 
 func (c *Client) heartBeat() {
@@ -330,7 +341,7 @@ func (c *Client) DeclarePublisher(streamName string, channelConfirmListener Publ
 	writeShort(b, int16(publisherReferenceSize))
 	writeString(b, streamName)
 	res := c.handleWrite(b.Bytes(), resp)
-	return producer, res
+	return producer, res.Err
 }
 
 func (c *Client) metaData(streams ...string) *StreamsMetadata {
@@ -353,7 +364,7 @@ func (c *Client) metaData(streams ...string) *StreamsMetadata {
 	}
 
 	err := c.handleWrite(b.Bytes(), resp)
-	if err != nil {
+	if err.Err != nil {
 		return nil
 	}
 
@@ -369,7 +380,7 @@ func (c *Client) BrokerLeader(stream string) (*Broker, error) {
 
 	streamMetadata := streamsMetadata.Get(stream)
 	if streamMetadata.responseCode != responseCodeOk {
-		return nil, fmt.Errorf("leader error for stream: %s, error:%s", stream, lookErrorCode(streamMetadata.responseCode))
+		return nil, lookErrorCode(streamMetadata.responseCode)
 	}
 	return streamMetadata.Leader, nil
 }
@@ -405,7 +416,7 @@ func (c *Client) DeclareStream(streamName string, options *StreamOptions) error
 		writeString(b, element)
 	}
 
-	return c.handleWrite(b.Bytes(), resp)
+	return c.handleWrite(b.Bytes(), resp).Err
 
 }
 
@@ -438,6 +449,12 @@ func (c *Client) DeclareSubscriber(ctx context.Context, streamName string,
 		// here we change the type since typeLastConsumed is not part of the protocol
 		options.Offset.typeOfs = typeOffset
 	}
+
+	// copy the option offset to the consumer offset
+	// the option.offset won't change ( in case we need to retrive the original configuration)
+	// consumer.current offset will be moved when reading
+	consumer.setCurrentOffset(options.Offset.offset)
+
 	resp := c.coordinator.NewResponse(commandSubscribe, streamName)
 	correlationId := resp.correlationid
 	var b = bytes.NewBuffer(make([]byte, 0, length+4))
@@ -467,7 +484,7 @@ func (c *Client) DeclareSubscriber(ctx context.Context, streamName string,
 				}
 
 			case data := <-consumer.response.data:
-				consumer.setOffset(data.(int64))
+				consumer.setCurrentOffset(data.(int64))
 
 			case messages := <-consumer.response.messages:
 				for _, message := range messages {
@@ -476,5 +493,5 @@ func (c *Client) DeclareSubscriber(ctx context.Context, streamName string,
 			}
 		}
 	}()
-	return consumer, err
+	return consumer, err.Err
 }

pkg/stream/client_test.go

@@ -23,7 +23,7 @@ var _ = AfterSuite(func() {
 	err := testEnvironment.Close()
 	Expect(err).NotTo(HaveOccurred())
 	time.Sleep(500 * time.Millisecond)
-	//Expect(testEnvironment.clientLocator.coordinator.ProducersCount()).To(Equal(0))
+	//Expect(testEnvironment.Coordinators()[0].ProducersCount()).To(Equal(0))
 	//Expect(testEnvironment.clientLocator.coordinator.ResponsesCount()).To(Equal(0))
 	//Expect(testEnvironment.clientLocator.coordinator.ConsumersCount()).To(Equal(0))
 })
@@ -112,8 +112,7 @@ var _ = Describe("Streaming testEnvironment", func() {
 		Expect(err).NotTo(HaveOccurred())
 		err = testEnvironment.DeclareStream(testStreamName, nil)
 		Expect(err).To(HaveOccurred())
-		Expect(fmt.Sprintf("%s", err)).
-			To(ContainSubstring("stream already exists"))
+		Expect(err).To(Equal(StreamAlreadyExists))
 		err = testEnvironment.DeleteStream(testStreamName)
 		Expect(err).NotTo(HaveOccurred())
 	})
@@ -127,8 +126,7 @@ var _ = Describe("Streaming testEnvironment", func() {
 				MaxLengthBytes: ByteCapacity{}.MB(100),
 			})
 		Expect(err).To(HaveOccurred())
-		Expect(fmt.Sprintf("%s", err)).
-			To(ContainSubstring("precondition failed"))
+		Expect(err).To(Equal(PreconditionFailed))
 		err = testEnvironment.DeleteStream(testStreamName)
 		Expect(err).NotTo(HaveOccurred())
 	})