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.
 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())) // or after 20 seconds
+```
+
+
 ### Handle Close
 Client provides an interface to handle the producer/consumer close.
 

examples/automaticOffsetTracking/automaticOffsetTracking.go

@@ -61,18 +61,18 @@ func main() {
 	consumerOffsetNumber, err := env.NewConsumer(streamName,
 		handleMessages,
 		stream.NewConsumerOptions().
-			SetConsumerName("my_consumer"). // set a consumerOffsetNumber name
-			SetAutoCommit().
-			SetAutoCommitStrategy(
-				stream.NewAutoCommitStrategy().
-					SetCountBeforeStorage(50).
-					SetFlushInterval(20*time.Second)))
+			// 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)
 
-	/// wait a bit just for demo and reset the counters
 	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))
@@ -82,7 +82,9 @@ func main() {
 		handleMessagesAfter,
 		stream.NewConsumerOptions().
 			SetConsumerName("my_consumer").                         // set a consumerOffsetNumber name
-			SetOffset(stream.OffsetSpecification{}.LastConsumed())) // with first() the the stream is loaded from the beginning
+			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 ")

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,31 +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 = consumer.messageCountBeforeStorage + 1
+						consumer.messageCountBeforeStorage += 1
 						if consumer.messageCountBeforeStorage >= consumer.options.autoCommitStrategy.messageCountBeforeStorage {
-							err := consumer.cacheStoreOffset()
-							if err != nil {
-								logs.LogError("message count before storage auto commit error : %s", err)
-							}
+							consumer.cacheStoreOffset()
 							consumer.messageCountBeforeStorage = 0
 						}
 					}
 				}
 
 			case <-time.After(consumer.options.autoCommitStrategy.flushInterval):
-				if consumer.options.autocommit {
-					err := consumer.cacheStoreOffset()
-					if err != nil {
-						logs.LogError("auto commit error: %s", err)
-					}
-				}
+				consumer.cacheStoreOffset()
+
 			}
 		}
 	}()

pkg/stream/consumer.go

@@ -65,13 +65,36 @@ func (consumer *Consumer) setCurrentOffset(offset int64) {
 	consumer.currentOffset = offset
 }
 
+func (consumer *Consumer) incCurrentOffset() {
+	consumer.mutex.Lock()
+	defer consumer.mutex.Unlock()
+	consumer.currentOffset += 1
+}
+
 func (consumer *Consumer) GetOffset() int64 {
 	consumer.mutex.Lock()
 	res := consumer.currentOffset
 	consumer.mutex.Unlock()
 	return res
 }
 
+func (consumer *Consumer) GetLastStoredOffset() int64 {
+	consumer.mutex.Lock()
+	res := consumer.lastStoredOffset
+	consumer.mutex.Unlock()
+	return res
+}
+
+func (consumer *Consumer) updateLastStoredOffset() bool {
+	consumer.mutex.Lock()
+	defer consumer.mutex.Unlock()
+	if consumer.lastStoredOffset < consumer.currentOffset {
+		consumer.lastStoredOffset = consumer.currentOffset
+		return true
+	}
+	return false
+}
+
 func (consumer *Consumer) NotifyClose() ChannelClose {
 	consumer.mutex.Lock()
 	defer consumer.mutex.Unlock()
@@ -102,8 +125,8 @@ func (ac *AutoCommitStrategy) SetFlushInterval(flushInterval time.Duration) *Aut
 
 func NewAutoCommitStrategy() *AutoCommitStrategy {
 	return &AutoCommitStrategy{
-		messageCountBeforeStorage: 20_000,
-		flushInterval:             10 * time.Second,
+		messageCountBeforeStorage: 10_000,
+		flushInterval:             5 * time.Second,
 	}
 }
 
@@ -128,13 +151,13 @@ func (c *ConsumerOptions) SetConsumerName(consumerName string) *ConsumerOptions
 	return c
 }
 
-func (c *ConsumerOptions) SetAutoCommit() *ConsumerOptions {
+func (c *ConsumerOptions) SetAutoCommit(autoCommitStrategy *AutoCommitStrategy) *ConsumerOptions {
 	c.autocommit = true
-	return c
-}
-
-func (c *ConsumerOptions) SetAutoCommitStrategy(autoCommitStrategy *AutoCommitStrategy) *ConsumerOptions {
-	c.autoCommitStrategy = autoCommitStrategy
+	if autoCommitStrategy == nil {
+		c.autoCommitStrategy = NewAutoCommitStrategy()
+	} else {
+		c.autoCommitStrategy = autoCommitStrategy
+	}
 	return c
 }
 
@@ -163,6 +186,8 @@ func (consumer *Consumer) Close() error {
 	if consumer.getStatus() == closed {
 		return AlreadyClosed
 	}
+	consumer.cacheStoreOffset()
+
 	consumer.setStatus(closed)
 	_, errGet := consumer.options.client.coordinator.GetConsumerById(consumer.ID)
 	if errGet != nil {
@@ -209,12 +234,13 @@ func (consumer *Consumer) Close() error {
 	return err.Err
 }
 
-func (consumer *Consumer) cacheStoreOffset() error {
-	if consumer.lastStoredOffset != consumer.GetOffset() {
-		consumer.lastStoredOffset = consumer.GetOffset()
-		return consumer.internalStoreOffset()
+func (consumer *Consumer) cacheStoreOffset() {
+	if consumer.options.autocommit {
+		err := consumer.internalStoreOffset()
+		if err != nil {
+			logs.LogError("cache Store Offset error : %s", err)
+		}
 	}
-	return nil
 }
 
 func (consumer *Consumer) StoreOffset() error {
@@ -224,17 +250,20 @@ func (consumer *Consumer) internalStoreOffset() error {
 	if consumer.options.streamName == "" {
 		return fmt.Errorf("stream Name can't be empty")
 	}
-	length := 2 + 2 + 2 + len(consumer.options.ConsumerName) + 2 +
-		len(consumer.options.streamName) + 8
-	var b = bytes.NewBuffer(make([]byte, 0, length+4))
-	writeProtocolHeader(b, length, commandStoreOffset)
 
-	writeString(b, consumer.options.ConsumerName)
-	writeString(b, consumer.options.streamName)
+	if consumer.updateLastStoredOffset() {
+		length := 2 + 2 + 2 + len(consumer.options.ConsumerName) + 2 +
+			len(consumer.options.streamName) + 8
+		var b = bytes.NewBuffer(make([]byte, 0, length+4))
+		writeProtocolHeader(b, length, commandStoreOffset)
 
-	writeLong(b, consumer.GetOffset())
-	return consumer.options.client.socket.writeAndFlush(b.Bytes())
+		writeString(b, consumer.options.ConsumerName)
+		writeString(b, consumer.options.streamName)
 
+		writeLong(b, consumer.GetOffset())
+		return consumer.options.client.socket.writeAndFlush(b.Bytes())
+	}
+	return nil
 }
 
 func (consumer *Consumer) QueryOffset() (int64, error) {