feat: automatic credit strategy is aligned with the java one (#440)

Author: hiimjako

pkg/stream/client.go

@@ -998,11 +998,19 @@ func (c *Client) declareSubscriber(streamName string,
 				if !ok {
 					return
 				}
-				for _, offMessage := range chunk.offsetMessages {
+
+				halfChunkSize := len(chunk.offsetMessages) / 2
+				for i, offMessage := range chunk.offsetMessages {
 					consumer.setCurrentOffset(offMessage.offset)
 					if canDispatch(offMessage) {
 						consumer.MessagesHandler(ConsumerContext{Consumer: consumer, chunkInfo: &chunk}, offMessage.message)
 					}
+
+					// when half of the chunk is reached ask for a credit
+					if halfChunkSize == i && consumer.options.CreditStrategy == AutomaticCreditStrategy {
+						c.credit(consumer.ID, 1)
+					}
+
 					if consumer.options.autocommit {
 						messageCountBeforeStorage := consumer.increaseMessageCountBeforeStorage()
 						if messageCountBeforeStorage >= consumer.options.autoCommitStrategy.messageCountBeforeStorage ||

pkg/stream/consumer_test.go

@@ -497,7 +497,8 @@ var _ = Describe("Streaming Consumers", func() {
 
 		// waiting the internal queue to fulfill
 		time.Sleep(time.Second)
-		Expect(len(consumer.chunkForConsumer)).To(Equal(credits))
+		// - 1 because the first message was consumed by the handler
+		Expect(len(consumer.chunkForConsumer)).To(Equal(credits - 1))
 
 		Expect(consumer.Close()).NotTo(HaveOccurred())
 	})

pkg/stream/coordinator.go

@@ -198,8 +198,9 @@ func (coordinator *Coordinator) NewConsumer(
 		lastStoredOffset:     -1, // because 0 is a valid value for the offset
 		isPromotedAsActive:   true,
 		lastAutoCommitStored: time.Now(),
-		chunkForConsumer:     make(chan chunkInfo, parameters.initialCredits),
-		onClose:              cleanUp,
+		// The +5 provides a small buffer to prevent the consumer loop from blocking.
+		chunkForConsumer: make(chan chunkInfo, parameters.initialCredits+5),
+		onClose:          cleanUp,
 	}
 
 	coordinator.consumers.Store(lastId, item)

pkg/stream/server_frame.go

@@ -409,10 +409,6 @@ func (c *Client) handleDeliver(r *bufio.Reader) {
 	chunk.offsetMessages = batchConsumingMessages
 	if consumer.getStatus() == open {
 		consumer.chunkForConsumer <- chunk
-		// request a credit for the next chunk
-		if consumer.options.CreditStrategy == AutomaticCreditStrategy {
-			c.credit(subscriptionId, 1)
-		}
 	} else {
 		logs.LogDebug("The consumer %s for the stream %s is closed during the chunk dispatching. "+
 			"Messages won't dispatched", consumer.GetName(), consumer.GetStreamName())