Query sequence

.gitignore

@@ -26,3 +26,4 @@ tls-gen/
 dist/
 
 perfTest/perfTest
+go.dev/

README.md

@@ -45,7 +45,7 @@ Go client for [RabbitMQ Stream Queues](https://github.com/rabbitmq/rabbitmq-serv
 ### Installing
 
 ```shell
- go get -u github.com/rabbitmq/rabbitmq-stream-go-client@v1.0.0-rc.13
+ go get -u github.com/rabbitmq/rabbitmq-stream-go-client
 ```
 
 imports:
@@ -296,6 +296,11 @@ https://blog.rabbitmq.com/posts/2021/07/rabbitmq-streams-message-deduplication/
 You can find a "Deduplication" example in the [examples](./examples/) directory. </br>
 Run it more than time, the messages count will be always 10.
 
+To retrieve the last sequence id for producer you can use:
+```
+publishingId, err := producer.GetLastPublishingId()
+```
+
 ### Sub Entries Batching
 
 The number of messages to put in a sub-entry. A sub-entry is one "slot" in a publishing frame, 

VERSION

@@ -1 +1 @@
-1.0.0-rc.13
+1.0.1-rc.1

examples/deduplication/deduplication.go

@@ -53,6 +53,13 @@ func main() {
 		}
 	}(chConfirm, producer)
 
+	// In case you need to know which is the last ID for the producer: GetLastPublishingId
+	lastPublishingId, err := producer.GetLastPublishingId()
+	CheckErr(err)
+	fmt.Printf("lastPublishingId: %d\n",
+		lastPublishingId,
+	)
+
 	data := make(map[int]string)
 	data[0] = "Piaggio"
 	data[1] = "Ferrari"

pkg/stream/client.go

@@ -527,7 +527,8 @@ func (c *Client) internalDeclarePublisher(streamName string, producer *Producer)
 	res := c.handleWrite(b.Bytes(), resp)
 
 	if publisherReferenceSize > 0 {
-		producer.sequence = c.queryPublisherSequence(producer.options.Name, streamName)
+		v, _ := c.queryPublisherSequence(producer.options.Name, streamName)
+		producer.sequence = v
 	}
 
 	return res
@@ -562,7 +563,7 @@ func (c *Client) metaData(streams ...string) *StreamsMetadata {
 	return data.(*StreamsMetadata)
 }
 
-func (c *Client) queryPublisherSequence(publisherReference string, stream string) int64 {
+func (c *Client) queryPublisherSequence(publisherReference string, stream string) (int64, error) {
 
 	length := 2 + 2 + 4 + 2 + len(publisherReference) + 2 + len(stream)
 	resp := c.coordinator.NewResponse(commandQueryPublisherSequence)
@@ -572,10 +573,13 @@ func (c *Client) queryPublisherSequence(publisherReference string, stream string
 
 	writeString(b, publisherReference)
 	writeString(b, stream)
-	c.handleWriteWithResponse(b.Bytes(), resp, false)
+	err := c.handleWriteWithResponse(b.Bytes(), resp, false)
 	sequence := <-resp.data
 	_ = c.coordinator.RemoveResponseById(resp.correlationid)
-	return sequence.(int64)
+	if err.Err != nil {
+		return 0, err.Err
+	}
+	return sequence.(int64), nil
 
 }
 

pkg/stream/constants.go

@@ -19,7 +19,7 @@ const (
 const initBufferPublishSize = 2 + 2 + 1 + 4
 
 const (
-	ClientVersion = "1.0.0-rc.13"
+	ClientVersion = "1.0.1-rc.1"
 
 	commandDeclarePublisher       = 1
 	commandPublish                = 2

pkg/stream/consumer_test.go

@@ -259,12 +259,13 @@ var _ = Describe("Streaming Consumers", func() {
 	})
 
 	It("Deduplication", func() {
-		producer, err := env.NewProducer(streamName, NewProducerOptions().SetProducerName("producer-ded"))
+		producerName := "producer-ded"
+		producer, err := env.NewProducer(streamName, NewProducerOptions().SetProducerName(producerName))
 		Expect(err).NotTo(HaveOccurred())
 		var arr []message.StreamMessage
 		for z := 0; z < 10; z++ {
 			m := amqp.NewMessage([]byte("test_" + strconv.Itoa(z)))
-			m.SetPublishingId(int64(z * 10))
+			m.SetPublishingId(int64(z * 10)) // id stored: the last one should be the same on QuerySequence
 			arr = append(arr, m)
 		}
 
@@ -300,6 +301,19 @@ var _ = Describe("Streaming Consumers", func() {
 			return atomic.LoadInt32(&messagesReceived)
 		}, 5*time.Second).Should(Equal(int32(10)),
 			"consumer should receive only 10 messages")
+
+		Eventually(func() int64 {
+			v, _ := env.QuerySequence(producerName, streamName)
+			return v
+		}, 5*time.Second).Should(Equal(int64(90)),
+			"QuerySequence should give the last id: 90")
+
+		Eventually(func() int64 {
+			v, _ := producer.GetLastPublishingId()
+			return v
+		}, 5*time.Second).Should(Equal(int64(90)),
+			"GetLastPublishingId should give the last id: 90")
+
 		Expect(producer.Close()).NotTo(HaveOccurred())
 		Expect(consumer.Close()).NotTo(HaveOccurred())
 	})

pkg/stream/enviroment.go

@@ -168,6 +168,22 @@ func (env *Environment) QueryOffset(consumerName string, streamName string) (int
 	return client.queryOffset(consumerName, streamName)
 }
 
+// QuerySequence gets the last id stored for a producer
+// you can also see producer.GetLastPublishingId() that is the easier way to get the last-id
+func (env *Environment) QuerySequence(publisherReference string, streamName string) (int64, error) {
+	client, err := env.newReconnectClient()
+	defer func(client *Client) {
+		err := client.Close()
+		if err != nil {
+			return
+		}
+	}(client)
+	if err != nil {
+		return 0, err
+	}
+	return client.queryPublisherSequence(publisherReference, streamName)
+}
+
 func (env *Environment) StreamMetaData(streamName string) (*StreamMetadata, error) {
 	client, err := env.newReconnectClient()
 	defer func(client *Client) {

pkg/stream/enviroment_test.go

@@ -261,6 +261,20 @@ var _ = Describe("Environment test", func() {
 
 	})
 
+	Describe("Validation Query Offset/Sequence", func() {
+
+		env, err := NewEnvironment(NewEnvironmentOptions())
+		Expect(err).NotTo(HaveOccurred())
+		_, err = env.QuerySequence("my_prod",
+			"Stream_Doesnt_exist")
+		Expect(err).To(HaveOccurred())
+
+		_, err = env.QueryOffset("my_cons",
+			"Stream_Doesnt_exist")
+		Expect(err).To(HaveOccurred())
+		Expect(env.Close()).NotTo(HaveOccurred())
+	})
+
 	Describe("Stream Existing/Meta data", func() {
 
 		env, err := NewEnvironment(NewEnvironmentOptions().SetPort(5552).

pkg/stream/producer.go

@@ -487,6 +487,9 @@ func (producer *Producer) FlushUnConfirmedMessages() {
 	producer.mutex.Unlock()
 }
 
+func (producer *Producer) GetLastPublishingId() (int64, error) {
+	return producer.options.client.queryPublisherSequence(producer.GetName(), producer.GetStreamName())
+}
 func (producer *Producer) Close() error {
 	if producer.getStatus() == closed {
 		return AlreadyClosed