handle close

Author: Gsantomaggio

pkg/stream/coordinator.go

@@ -43,10 +43,10 @@ func (coordinator *Coordinator) NewProducer(
 	parameters *ProducerOptions) (*Producer, error) {
 	coordinator.mutex.Lock()
 	defer coordinator.mutex.Unlock()
-	size := 10000
-	if parameters != nil {
-		size = parameters.QueueSize
-	}
+	////size := 1
+	//if parameters != nil {
+	//	size = parameters.QueueSize
+	//}
 
 	var lastId, err = coordinator.getNextProducerItem()
 	if err != nil {
@@ -57,7 +57,7 @@ func (coordinator *Coordinator) NewProducer(
 		mutex:               &sync.Mutex{},
 		unConfirmedMessages: map[int64]*ConfirmationStatus{},
 		status:              open,
-		messageSequenceCh:   make(chan messageSequence, size),
+		messageSequenceCh:   make(chan messageSequence, 0),
 		pendingMessages: pendingMessagesSequence{
 			messages: make([]messageSequence, 0),
 			size:     initBufferPublishSize,
@@ -93,6 +93,7 @@ func (coordinator *Coordinator) RemoveProducerById(id uint8, reason Event) error
 	for producer.lenUnConfirmed() > 0 && tentatives < 3 {
 		time.Sleep(500 * time.Millisecond)
 		tentatives++
+		//logs.LogInfo("%d %d", producer.lenUnConfirmed(), len(producer.pendingMessages.messages))
 	}
 	producer.FlushUnConfirmedMessages()
 

pkg/stream/producer.go

@@ -401,6 +401,7 @@ func (producer *Producer) aggregateEntities(msgs []messageSequence, size int, co
 
 /// the producer id is always the producer.GetID(). This function is needed only for testing
 // some condition, like simulate publish error, see
+var count int32
 func (producer *Producer) internalBatchSendProdId(messagesSequence []messageSequence, producerID uint8) error {
 	producer.options.client.socket.mutex.Lock()
 	defer producer.options.client.socket.mutex.Unlock()
@@ -434,6 +435,7 @@ func (producer *Producer) internalBatchSendProdId(messagesSequence []messageSequ
 	writeBProtocolHeader(producer.options.client.socket.writer, length, commandPublish)
 	writeBByte(producer.options.client.socket.writer, producerID)
 	numberOfMessages := len(messagesSequence)
+	atomic.AddInt32(&count, int32(numberOfMessages))
 	numberOfMessages = numberOfMessages / producer.options.SubEntrySize
 	if len(messagesSequence)%producer.options.SubEntrySize != 0 {
 		numberOfMessages += 1
@@ -481,6 +483,9 @@ func (producer *Producer) Close() error {
 		return AlreadyClosed
 	}
 	producer.setStatus(closed)
+	time.Sleep(time.Duration(producer.options.BatchPublishingDelay) * time.Millisecond)
+	producer.sendBufferedMessages()
+
 	if !producer.options.client.socket.isOpen() {
 		return fmt.Errorf("tcp connection is closed")
 	}