Expose query Offset to the Environment level (#136)

* Expose query Offset to the Environment level

closes #132

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

* Change version

Author: Gsantomaggio

README.md

@@ -31,6 +31,7 @@ Go client for [RabbitMQ Stream Queues](https://github.com/rabbitmq/rabbitmq-serv
     * [Consume messages](#consume-messages)
         * [Manual Track Offset](#manual-track-offset)
         * [Automatic Track Offset](#automatic-track-offset)
+        * [Get consumer Offset](#get-consumer-offset)
     * [Handle Close](#handle-close)
 - [Performance test tool](#performance-test-tool)
     * [Performance test tool Docker](#performance-test-tool-docker)
@@ -44,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-rc12
+go get -u github.com/rabbitmq/rabbitmq-stream-go-client
 ```
 
 imports:
@@ -423,6 +424,15 @@ stream.NewConsumerOptions().
 
 See also "Automatic Offset Tracking" example in the [examples](./examples/) directory
 
+### Get consumer offset
+
+It is possible to query the consumer offset using:
+```golang
+offset, err := env.QueryOffset("consumer_name", "streamName")
+```
+An error is returned if the offset doesn't exist.
+
+
 ### Handle Close
 Client provides an interface to handle the producer/consumer close.
 

VERSION

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

change_version

@@ -5,8 +5,8 @@ go fmt ./...
 rm pkg/stream/constants.go-e
 
 
-sed -i -e "s/.*github.com\/rabbitmq\/rabbitmq-stream-go-client@v*.*/go get -u github.com\/rabbitmq\/rabbitmq-stream-go-client@v$1/" README.md
+# sed -i -e "s/.*github.com\/rabbitmq\/rabbitmq-stream-go-client@v*.*/go get -u github.com\/rabbitmq\/rabbitmq-stream-go-client@v$1/" README.md
 go fmt ./...
-rm README.md-e
+# rm README.md-e
 
 git add VERSION pkg/stream/constants.go README.md

examples/getting_started.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"github.com/google/uuid"
 	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
-	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/logs"
 	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
 	"os"
 	"strconv"
@@ -43,7 +42,7 @@ func main() {
 	reader := bufio.NewReader(os.Stdin)
 	// Set log level, not mandatory by default is INFO
 	// you cn set DEBUG for more information
-   //  stream.SetLevelInfo(logs.DEBUG)
+	// stream.SetLevelInfo(logs.DEBUG)
 
 	fmt.Println("Getting started with Streaming client for RabbitMQ")
 	fmt.Println("Connecting to RabbitMQ streaming ...")

pkg/stream/constants.go

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

pkg/stream/consumer_test.go

@@ -205,7 +205,6 @@ var _ = Describe("Streaming Consumers", func() {
 			Eventually(func() int64 { return consumer.GetLastStoredOffset() }, 5*time.Second).Should(Equal(int64(99)),
 				"Offset should be 99")
 			Expect(consumer.Close()).NotTo(HaveOccurred())
-
 		})
 
 		It("automatically commit by number/time", func() {
@@ -226,7 +225,7 @@ var _ = Describe("Streaming Consumers", func() {
 				"Offset should be 100")
 			Expect(consumer.Close()).NotTo(HaveOccurred())
 			/// When the consumer is closed, it has to save the offset
-			// so  the last offest has to be 105
+			// so  the last offset has to be 105
 			Eventually(func() int64 {
 				return consumer.GetLastStoredOffset()
 			}, 5*time.Second).Should(Equal(int64(105)),
@@ -318,6 +317,9 @@ var _ = Describe("Streaming Consumers", func() {
 	It("Subscribe/Unsubscribe count messages manual store", func() {
 		producer, err := env.NewProducer(streamName, nil)
 		Expect(err).NotTo(HaveOccurred())
+		// the offset doesn't exist (yet) here for the consumer test
+		_, err = env.QueryOffset("consumer_test", streamName)
+		Expect(err).To(HaveOccurred())
 
 		Expect(producer.BatchSend(CreateArrayMessagesForTesting(107))).
 			NotTo(HaveOccurred())
@@ -341,6 +343,13 @@ var _ = Describe("Streaming Consumers", func() {
 		}, 5*time.Second).Should(Equal(int64(107)),
 			"Offset should be 107")
 
+		offset, err := env.QueryOffset("consumer_test", streamName)
+		Expect(err).NotTo(HaveOccurred())
+		Eventually(func() int64 {
+			return offset
+		}, 5*time.Second).Should(Equal(int64(107)),
+			"Offset should be 107")
+
 		err = consumer.Close()
 		Expect(err).NotTo(HaveOccurred())
 

pkg/stream/enviroment.go

@@ -154,6 +154,20 @@ func (env *Environment) StreamExists(streamName string) (bool, error) {
 	return client.StreamExists(streamName), nil
 }
 
+func (env *Environment) QueryOffset(consumerName 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.queryOffset(consumerName, streamName)
+}
+
 func (env *Environment) StreamMetaData(streamName string) (*StreamMetadata, error) {
 	client, err := env.newReconnectClient()
 	defer func(client *Client) {