Auto track offest (#108)

* Fixes #102

Author: Gsantomaggio

README.md

@@ -29,7 +29,8 @@ Go client for [RabbitMQ Stream Queues](https://github.com/rabbitmq/rabbitmq-serv
         * [Sub Entries Batching](#sub-entries-batching)
         * [HA producer - Experimental](#ha-producer-experimental)
     * [Consume messages](#consume-messages)
-        * [Track Offset](#track-offset)
+        * [Manual Track Offset](#manual-track-offset)
+        * [Automatic Track Offset](#automatic-track-offset)
     * [Handle Close](#handle-close)
 - [Performance test tool](#performance-test-tool)
     * [Performance test tool Docker](#performance-test-tool-docker)
@@ -336,7 +337,7 @@ consumer, err := env.NewConsumer(
 		handleMessages,
 		....
 ```
-management UI
+
 With `ConsumerOptions` it is possible to customize the consumer behaviour.
 ```golang
   stream.NewConsumerOptions().
@@ -346,7 +347,7 @@ With `ConsumerOptions` it is possible to customize the consumer behaviour.
 See also "Offset Start" example in the [examples](./examples/) directory
 
 
-### Track Offset
+### Manual Track Offset
 The server can store the offset given a consumer, in this way:
 ```golang
 handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
@@ -359,10 +360,48 @@ consumer, err := env.NewConsumer(
 stream.NewConsumerOptions().
 			SetConsumerName("my_consumer"). <------ 
 ```
+A consumer must have a name to be able to store offsets. <br>
 Note: *AVOID to store the offset for each single message, it will reduce the performances*
 
 See also "Offset Tracking" example in the [examples](./examples/) directory
 
+### Automatic Track Offset
+
+The following snippet shows how to enable automatic tracking with the defaults:
+```golang
+stream.NewConsumerOptions().
+			SetConsumerName("my_consumer").
+			SetAutoCommit(stream.NewAutoCommitStrategy() ...
+```
+`nil` is also a valid value. Default values will be used
+
+```golang
+stream.NewConsumerOptions().
+			SetConsumerName("my_consumer").
+			SetAutoCommit(nil) ...
+```
+Set the consumer name (mandatory for offset tracking) </br>
+
+The automatic tracking strategy has the following available settings:
+- message count before storage: the client will store the offset after the specified number of messages, </br>
+right after the execution of the message handler. The default is every 10,000 messages.
+
+- flush interval: the client will make sure to store the last received offset at the specified interval. </br>
+This avoids having pending, not stored offsets in case of inactivity.  The default is 5 seconds.
+
+Those settings are configurable, as shown in the following snippet:
+```golang
+stream.NewConsumerOptions().
+	// set a consumerOffsetNumber name
+	SetConsumerName("my_consumer").
+	SetAutoCommit(stream.NewAutoCommitStrategy().
+		SetCountBeforeStorage(50). // store each 50 messages stores 
+		SetFlushInterval(10*time.Second)). // store each 10 seconds
+	SetOffset(stream.OffsetSpecification{}.First())) 
+```
+
+See also "Automatic Offset Tracking" example in the [examples](./examples/) directory
+
 ### Handle Close
 Client provides an interface to handle the producer/consumer close.
 

examples/README.md

@@ -3,7 +3,8 @@ Stream examples
 
  - [Getting started](./getting_started.go). A good point to start.
  - [Offset Start](./offsetStart/offset.go). How to set different points to start consuming
- - [Offset Tracking](./offsetTracking/offsetTracking.go). How to store the consumer position server side
+ - [Offset Tracking](./offsetTracking/offsetTracking.go). Manually store the consumer offset 
+ - [Automatic Offset Tracking](./automaticOffsetTracking/automaticOffsetTracking.go). Automatic store the consumer offset
  - [Getting started TLS](./tls/getting_started_tls.go). A TLS example.
  - [HA Producer](./haProducer/producer.go). HA producer example (Still experimental)
  - [Deduplication](./deduplication/deduplication.go). deduplication example, run it more than one time, and the records <br />

examples/automaticOffsetTracking/automaticOffsetTracking.go

@@ -0,0 +1,101 @@
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"github.com/google/uuid"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
+	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
+	"os"
+	"strconv"
+	"sync/atomic"
+	"time"
+)
+
+func CheckErr(err error) {
+	if err != nil {
+		fmt.Printf("%s ", err)
+		os.Exit(1)
+	}
+}
+
+func main() {
+	reader := bufio.NewReader(os.Stdin)
+
+	fmt.Println("Automatic Offset tracking example")
+	fmt.Println("Connecting to RabbitMQ streaming ...")
+
+	env, err := stream.NewEnvironment(
+		stream.NewEnvironmentOptions().
+			SetHost("localhost").
+			SetPort(5552).
+			SetUser("guest").
+			SetPassword("guest"))
+	CheckErr(err)
+	streamName := uuid.New().String()
+	err = env.DeclareStream(streamName,
+		&stream.StreamOptions{
+			MaxLengthBytes: stream.ByteCapacity{}.GB(2),
+		},
+	)
+
+	CheckErr(err)
+
+	producer, err := env.NewProducer(streamName, nil)
+	CheckErr(err)
+
+	go func() {
+		for i := 0; i < 220; i++ {
+			err := producer.Send(amqp.NewMessage([]byte("hello_world_" + strconv.Itoa(i))))
+			CheckErr(err)
+		}
+	}()
+
+	var counter int32
+	handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
+		if atomic.AddInt32(&counter, 1)%20 == 0 {
+			fmt.Printf("messages consumed with auto commit: %d \n ", atomic.LoadInt32(&counter))
+		}
+	}
+
+	consumerOffsetNumber, err := env.NewConsumer(streamName,
+		handleMessages,
+		stream.NewConsumerOptions().
+			// set a consumerOffsetNumber name
+			SetConsumerName("my_consumer").
+			// nil is also a valid value. Default values will be used
+			SetAutoCommit(stream.NewAutoCommitStrategy().
+										SetCountBeforeStorage(50). // each 50 messages stores the index
+										SetFlushInterval(20*time.Second)).
+			SetOffset(stream.OffsetSpecification{}.First())) // or after 20 seconds
+	CheckErr(err)
+
+	time.Sleep(2 * time.Second)
+	atomic.StoreInt32(&counter, 0)
+	// so here we consume only 20 messages
+	handleMessagesAfter := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
+		if atomic.AddInt32(&counter, 1)%20 == 0 {
+			fmt.Printf("messages consumed after: %d \n ", atomic.LoadInt32(&counter))
+		}
+	}
+	consumerNext, err := env.NewConsumer(streamName,
+		handleMessagesAfter,
+		stream.NewConsumerOptions().
+			SetConsumerName("my_consumer").                         // set a consumerOffsetNumber name
+			SetOffset(stream.OffsetSpecification{}.LastConsumed())) // With last consumed we point to the last saved.
+	// in this case will be 200. So it will consume 20
+	//messages
+	CheckErr(err)
+
+	fmt.Println("Press any key to stop ")
+	_, _ = reader.ReadString('\n')
+	err = producer.Close()
+	CheckErr(err)
+	err = consumerOffsetNumber.Close()
+	CheckErr(err)
+	err = consumerNext.Close()
+	CheckErr(err)
+	err = env.DeleteStream(streamName)
+	CheckErr(err)
+
+}

pkg/stream/client.go

@@ -683,6 +683,14 @@ func (c *Client) DeclareSubscriber(streamName string,
 		return nil, fmt.Errorf("specify a valid Offset")
 	}
 
+	if options.autoCommitStrategy.flushInterval < 1*time.Second {
+		return nil, fmt.Errorf("flush internal must be bigger than one second")
+	}
+
+	if options.autoCommitStrategy.messageCountBeforeStorage < 1 {
+		return nil, fmt.Errorf("message count before storage be bigger than one")
+	}
+
 	options.client = c
 	options.streamName = streamName
 	consumer := c.coordinator.NewConsumer(messagesHandler, options)
@@ -741,14 +749,22 @@ func (c *Client) DeclareSubscriber(streamName string,
 					return
 				}
 
-			case data := <-consumer.response.data:
-				consumer.setCurrentOffset(data.(int64))
-
-			case messages := <-consumer.response.messages:
-				for _, message := range messages {
+			case offsetMessages := <-consumer.response.offsetMessages:
+				for _, message := range offsetMessages.messages {
+					consumer.incCurrentOffset()
 					consumer.MessagesHandler(ConsumerContext{Consumer: consumer}, message)
+					if consumer.options.autocommit {
+						consumer.messageCountBeforeStorage += 1
+						if consumer.messageCountBeforeStorage >= consumer.options.autoCommitStrategy.messageCountBeforeStorage {
+							consumer.cacheStoreOffset()
+							consumer.messageCountBeforeStorage = 0
+						}
+					}
 				}
 
+			case <-time.After(consumer.options.autoCommitStrategy.flushInterval):
+				consumer.cacheStoreOffset()
+
 			}
 		}
 	}()