wip

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

Author: Gsantomaggio

pkg/stream/blocking_queue.go

@@ -10,15 +10,18 @@ import (
 var ErrBlockingQueueStopped = errors.New("blocking queue stopped")
 
 type BlockingQueue[T any] struct {
-	queue  chan T
-	status int32
+	queue      chan T
+	status     int32
+	capacity   int
+	lastUpdate time.Time
 }
 
 // NewBlockingQueue initializes a new BlockingQueue with the given capacity
 func NewBlockingQueue[T any](capacity int) *BlockingQueue[T] {
 	return &BlockingQueue[T]{
-		queue:  make(chan T, capacity),
-		status: 0,
+		queue:    make(chan T, capacity),
+		capacity: capacity,
+		status:   0,
 	}
 }
 
@@ -27,6 +30,7 @@ func (bq *BlockingQueue[T]) Enqueue(item T) error {
 	if bq.IsStopped() {
 		return ErrBlockingQueueStopped
 	}
+	bq.lastUpdate = time.Now()
 	bq.queue <- item
 
 	return nil
@@ -59,6 +63,13 @@ func (bq *BlockingQueue[T]) IsEmpty() bool {
 	return len(bq.queue) == 0
 }
 
+func (bq *BlockingQueue[T]) IsReadyToSend() bool {
+	if bq.lastUpdate.IsZero() {
+		return true
+	}
+	return time.Since(bq.lastUpdate) > 10*time.Millisecond && len(bq.queue) == 0
+}
+
 // Stop stops the queue from accepting new items
 // but allows some pending items.
 // Stop is different from Close in that it allows the

pkg/stream/producer.go

@@ -287,6 +287,9 @@ func (producer *Producer) closeConfirmationStatus() {
 func (producer *Producer) processPendingSequencesQueue() {
 
 	maxFrame := producer.options.client.getTuneState().requestedMaxFrameSize
+	var avarage int32
+	iterations := 0
+
 	// the buffer is initialized with the size of the header
 	sequenceToSend := make([]*messageSequence, 0)
 	go func() {
@@ -316,8 +319,16 @@ func (producer *Producer) processPendingSequencesQueue() {
 
 			// if producer.pendingSequencesQueue.IsEmpty() means that the queue is empty so the producer is not sending
 			// the messages during the checks of the buffer. In this case
-			if producer.pendingSequencesQueue.IsEmpty() || len(sequenceToSend) >= producer.options.BatchSize {
+			if producer.pendingSequencesQueue.IsReadyToSend() || len(sequenceToSend) >= producer.options.BatchSize {
 				if len(sequenceToSend) > 0 {
+					avarage = atomic.AddInt32(&avarage, int32(len(sequenceToSend)))
+					iterations++
+					if iterations > 10000 {
+						logs.LogInfo("producer %d avarage: %d", producer.id, avarage/int32(iterations))
+						avarage = 0
+						iterations = 0
+					}
+
 					lastError = producer.internalBatchSend(sequenceToSend)
 					sequenceToSend = sequenceToSend[:0]
 					totalBufferToSend += initBufferPublishSize