add function to process the messages

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

Author: Gsantomaggio

pkg/stream/blocking_queue.go

@@ -55,6 +55,20 @@ func (bq *BlockingQueue[T]) Dequeue(timeout time.Duration) T {
 	}
 }
 
+func (bq *BlockingQueue[T]) Process(f func(T)) {
+	isActive := true
+	for isActive {
+		select {
+		case item, ok := <-bq.queue:
+			if !ok {
+				isActive = false
+				return
+			}
+			f(item)
+		}
+	}
+}
+
 func (bq *BlockingQueue[T]) Size() int {
 	return len(bq.queue)
 }

pkg/stream/producer.go

@@ -289,20 +289,11 @@ func (producer *Producer) processPendingSequencesQueue() {
 	maxFrame := producer.options.client.getTuneState().requestedMaxFrameSize
 	var avarage = 0
 	iterations := 0
-
-	// the buffer is initialized with the size of the header
-	sequenceToSend := make([]*messageSequence, 0)
 	go func() {
+		sequenceToSend := make([]*messageSequence, 0)
 		totalBufferToSend := initBufferPublishSize
-		for {
+		producer.pendingSequencesQueue.Process(func(msg *messageSequence) {
 			var lastError error
-			// the dequeue is blocking with a timeout of 500ms
-			// as soon as a message is available the Dequeue will be unblocked
-			msg := producer.pendingSequencesQueue.Dequeue(time.Millisecond * 500)
-			if producer.pendingSequencesQueue.IsStopped() {
-				break
-			}
-
 			if msg != nil {
 				// There is something in the queue.Checks the buffer is still less than the maxFrame
 				totalBufferToSend += msg.unCompressedSize
@@ -313,18 +304,14 @@ func (producer *Producer) processPendingSequencesQueue() {
 					sequenceToSend = sequenceToSend[:0]
 					totalBufferToSend = initBufferPublishSize
 				}
-
 				sequenceToSend = append(sequenceToSend, msg)
 			}
-
-			// 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.IsReadyToSend() || len(sequenceToSend) >= producer.options.BatchSize {
 				if len(sequenceToSend) > 0 {
 					avarage += len(sequenceToSend)
 					iterations++
 					if iterations > 10000 {
-						logs.LogInfo("producer %d avarage: %d", producer.id, avarage/iterations)
+						logs.LogInfo("producer %d average: %d", producer.id, avarage/iterations)
 						avarage = 0
 						iterations = 0
 					}
@@ -337,9 +324,58 @@ func (producer *Producer) processPendingSequencesQueue() {
 			if lastError != nil {
 				logs.LogError("error during sending messages: %s", lastError)
 			}
-		}
-		logs.LogDebug("producer %d processPendingSequencesQueue closed", producer.id)
+
+		})
 	}()
+	logs.LogDebug("producer %d processPendingSequencesQueue closed", producer.id)
+	// the buffer is initialized with the size of the header
+
+	//	for {
+	//		var lastError error
+	//		// the dequeue is blocking with a timeout of 500ms
+	//		// as soon as a message is available the Dequeue will be unblocked
+	//		msg := producer.pendingSequencesQueue.Dequeue(time.Millisecond * 500)
+	//		if producer.pendingSequencesQueue.IsStopped() {
+	//			break
+	//		}
+	//
+	//		if msg != nil {
+	//			// There is something in the queue.Checks the buffer is still less than the maxFrame
+	//			totalBufferToSend += msg.unCompressedSize
+	//			if totalBufferToSend > maxFrame {
+	//				// if the totalBufferToSend is greater than the requestedMaxFrameSize
+	//				// the producer sends the messages and reset the buffer
+	//				lastError = producer.internalBatchSend(sequenceToSend)
+	//				sequenceToSend = sequenceToSend[:0]
+	//				totalBufferToSend = initBufferPublishSize
+	//			}
+	//
+	//			sequenceToSend = append(sequenceToSend, msg)
+	//		}
+	//
+	//		// 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.IsReadyToSend() || len(sequenceToSend) >= producer.options.BatchSize {
+	//			if len(sequenceToSend) > 0 {
+	//				avarage += len(sequenceToSend)
+	//				iterations++
+	//				if iterations > 10000 {
+	//					logs.LogInfo("producer %d average: %d", producer.id, avarage/iterations)
+	//					avarage = 0
+	//					iterations = 0
+	//				}
+	//
+	//				lastError = producer.internalBatchSend(sequenceToSend)
+	//				sequenceToSend = sequenceToSend[:0]
+	//				totalBufferToSend += initBufferPublishSize
+	//			}
+	//		}
+	//		if lastError != nil {
+	//			logs.LogError("error during sending messages: %s", lastError)
+	//		}
+	//	}
+	//	logs.LogDebug("producer %d processPendingSequencesQueue closed", producer.id)
+	//}()
 }
 
 func (producer *Producer) assignPublishingID(message message.StreamMessage) int64 {