remove loop

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

pkg/stream/aggregation_test.go

@@ -18,9 +18,8 @@ var _ = Describe("Compression algorithms", func() {
 		}
 
 		message := &messageSequence{
-			messageBytes:     messagePayload,
-			unCompressedSize: len(messagePayload),
-			publishingId:     0,
+			messageBytes: messagePayload,
+			publishingId: 0,
 		}
 
 		entries = &subEntries{

pkg/stream/client.go

@@ -432,7 +432,7 @@ func (c *Client) heartBeat() {
 			tickerHeartbeat.Stop()
 			return
 		case <-tickerHeartbeat.C:
-			for c.socket.isOpen() {
+			if c.socket.isOpen() {
 				if time.Since(c.getLastHeartBeat()) > time.Duration(c.tuneState.requestedHeartbeat)*time.Second {
 					v := atomic.AddInt32(&heartBeatMissed, 1)
 					logs.LogWarn("Missing heart beat: %d", v)

pkg/stream/consumer.go

@@ -337,10 +337,10 @@ func (consumer *Consumer) close(reason Event) error {
 		consumer.closeHandler = nil
 	}
 
-	if consumer.chunkForConsumer != nil {
-		close(consumer.chunkForConsumer)
-		consumer.chunkForConsumer = nil
-	}
+	//if consumer.chunkForConsumer != nil {
+	close(consumer.chunkForConsumer)
+	//consumer.chunkForConsumer = nil
+	//}
 
 	if consumer.response.data != nil {
 		close(consumer.response.data)

pkg/stream/coordinator.go

@@ -73,11 +73,6 @@ func (coordinator *Coordinator) NewProducer(
 		status:                    open,
 		pendingSequencesQueue:     NewBlockingQueue[*messageSequence](dynSize),
 		confirmMutex:              &sync.Mutex{},
-		pool: &sync.Pool{
-			New: func() any {
-				return &messageSequence{}
-			},
-		},
 	}
 	coordinator.producers[lastId] = producer
 	return producer, err

pkg/stream/producer.go

@@ -73,7 +73,6 @@ type Producer struct {
 	publishConfirmation chan []*ConfirmationStatus
 
 	pendingSequencesQueue *BlockingQueue[*messageSequence]
-	pool                  *sync.Pool
 }
 
 type FilterValue func(message message.StreamMessage) string
@@ -296,7 +295,6 @@ func (producer *Producer) processPendingSequencesQueue() {
 		var lastError error
 		for {
 			select {
-
 			case msg, ok := <-producer.pendingSequencesQueue.GetChannel():
 				{
 					if !ok {
@@ -379,12 +377,12 @@ func (producer *Producer) fromMessageToMessageSequence(streamMessage message.Str
 	if producer.options.IsFilterEnabled() {
 		filterValue = producer.options.Filter.FilterValue(streamMessage)
 	}
-	fromPool := producer.pool.Get().(*messageSequence)
-	fromPool.messageBytes = marshalBinary
-	fromPool.publishingId = seq
-	fromPool.filterValue = filterValue
+	msqSeq := &messageSequence{}
+	msqSeq.messageBytes = marshalBinary
+	msqSeq.publishingId = seq
+	msqSeq.filterValue = filterValue
 
-	return fromPool, nil
+	return msqSeq, nil
 
 }
 
@@ -403,7 +401,6 @@ func (producer *Producer) Send(streamMessage message.StreamMessage) error {
 	if len(messageSeq.messageBytes) > defaultMaxFrameSize {
 		tooLarge := producer.unConfirmed.extractWithError(messageSeq.publishingId, responseCodeFrameTooLarge)
 		producer.sendConfirmationStatus([]*ConfirmationStatus{tooLarge})
-		producer.pool.Put(messageSeq)
 		return FrameTooLarge
 	}
 
@@ -443,7 +440,6 @@ func (producer *Producer) BatchSend(batchMessages []message.StreamMessage) error
 		for _, msg := range messagesSequence {
 			m := producer.unConfirmed.extractWithError(msg.publishingId, responseCodeFrameTooLarge)
 			producer.sendConfirmationStatus([]*ConfirmationStatus{m})
-			producer.pool.Put(msg)
 		}
 		return FrameTooLarge
 	}
@@ -457,13 +453,6 @@ func (producer *Producer) GetID() uint8 {
 }
 
 func (producer *Producer) internalBatchSend(messagesSequence []*messageSequence) error {
-	// remove form pool
-	defer func() {
-		for _, m := range messagesSequence {
-			producer.pool.Put(m)
-		}
-	}()
-
 	return producer.internalBatchSendProdId(messagesSequence, producer.GetID())
 }
 
@@ -535,8 +524,8 @@ func (producer *Producer) aggregateEntities(msgs []*messageSequence, size int, c
 // / the producer id is always the producer.GetID(). This function is needed only for testing
 // some condition, like simulate publish error.
 func (producer *Producer) internalBatchSendProdId(messagesSequence []*messageSequence, producerID uint8) error {
-	//producer.options.client.socket.mutexMessageMap.Lock()
-	//defer producer.options.client.socket.mutexMessageMap.Unlock()
+	producer.options.client.socket.mutex.Lock()
+	defer producer.options.client.socket.mutex.Unlock()
 	if producer.getStatus() == closed {
 		return fmt.Errorf("producer id: %d closed", producer.id)
 	}

pkg/stream/producer_test.go

@@ -315,7 +315,7 @@ var _ = Describe("Streaming Producers", func() {
 
 	})
 
-	It("Smart Send/Close", Focus, func() {
+	It("Smart Send/Close", func() {
 		producer, err := testEnvironment.NewProducer(testProducerStream, nil)
 		Expect(err).NotTo(HaveOccurred())
 		var messagesReceived int32
@@ -570,17 +570,15 @@ var _ = Describe("Streaming Producers", func() {
 		for i := 0; i < 1; i++ {
 			s := make([]byte, 50)
 			messagesSequence[i] = &messageSequence{
-				messageBytes:     s,
-				unCompressedSize: len(s),
+				messageBytes: s,
 			}
 		}
 
 		msg := amqp.NewMessage([]byte("test"))
 		msg.SetPublishingId(1)
 		messageBytes, _ := msg.MarshalBinary()
 		messagesSequence[0] = &messageSequence{
-			messageBytes:     messageBytes,
-			unCompressedSize: len(messageBytes),
+			messageBytes: messageBytes,
 		}
 
 		// 200 producer ID doesn't exist
@@ -659,8 +657,7 @@ var _ = Describe("Streaming Producers", func() {
 		for i := 0; i < 201; i++ {
 			s := make([]byte, 50)
 			messagesSequence[i] = &messageSequence{
-				messageBytes:     s,
-				unCompressedSize: len(s),
+				messageBytes: s,
 			}
 		}
 
@@ -676,8 +673,7 @@ var _ = Describe("Streaming Producers", func() {
 
 			s := make([]byte, 50)
 			messagesSequence[i] = &messageSequence{
-				messageBytes:     s,
-				unCompressedSize: len(s),
+				messageBytes: s,
 			}
 		}
 
@@ -692,8 +688,7 @@ var _ = Describe("Streaming Producers", func() {
 		for i := 0; i < 1; i++ {
 			s := make([]byte, 50)
 			messagesSequence[i] = &messageSequence{
-				messageBytes:     s,
-				unCompressedSize: len(s),
+				messageBytes: s,
 			}
 		}
 
@@ -708,8 +703,7 @@ var _ = Describe("Streaming Producers", func() {
 		for i := 0; i < 1000; i++ {
 			s := make([]byte, 50)
 			messagesSequence[i] = &messageSequence{
-				messageBytes:     s,
-				unCompressedSize: len(s),
+				messageBytes: s,
 			}
 		}
 
@@ -724,8 +718,7 @@ var _ = Describe("Streaming Producers", func() {
 		for i := 0; i < 14; i++ {
 			s := make([]byte, 50)
 			messagesSequence[i] = &messageSequence{
-				messageBytes:     s,
-				unCompressedSize: len(s),
+				messageBytes: s,
 			}
 		}