remove correlation id con commit offest

rabbitmq/rabbitmq-server#3097

Author: Gsantomaggio

examples/getting_started.go

@@ -29,9 +29,14 @@ func CreateArrayMessagesForTesting(bacthMessages int) []*amqp.Message {
 
 func handlePublishConfirm(confirms stream.ChannelPublishConfirm) {
 	go func() {
-		for messagesIds := range confirms {
-			for _, m := range messagesIds {
-				fmt.Printf("Confirmed %s message \n  ", m.Message.Data)
+		for confirmed := range confirms {
+			for _, msg := range confirmed {
+				if msg.Confirmed {
+					fmt.Printf("message %s stored \n  ", msg.Message.Data)
+				} else {
+					fmt.Printf("message %s failed \n  ", msg.Message.Data)
+				}
+
 			}
 		}
 	}()

pkg/stream/consumer.go

@@ -154,11 +154,10 @@ func (consumer *Consumer) Commit() error {
 	if consumer.options.streamName == "" {
 		return fmt.Errorf("stream Name can't be empty")
 	}
-	length := 2 + 2 + 4 + 2 + len(consumer.options.ConsumerName) + 2 +
+	length := 2 + 2 + 2 + len(consumer.options.ConsumerName) + 2 +
 		len(consumer.options.streamName) + 8
 	var b = bytes.NewBuffer(make([]byte, 0, length+4))
-	writeProtocolHeader(b, length, commandCommitOffset,
-		0) // correlation ID not used yet, may be used if commit offset has a confirm
+	writeProtocolHeader(b, length, commandCommitOffset) // correlation ID not used yet, may be used if commit offset has a confirm
 
 	writeString(b, consumer.options.ConsumerName)
 	writeString(b, consumer.options.streamName)

pkg/stream/coordinator.go

@@ -81,15 +81,15 @@ func (coordinator *Coordinator) RemoveProducerById(id uint8, reason Event) error
 		time.Sleep(200 * time.Millisecond)
 		tentatives++
 	}
-	//producer.mutex.Lock()
-	//if producer.publishConfirm != nil {
-	//	for _, message := range producer.unConfirmedMessages {
-	//		message.Confirmed = false
-	//		producer.publishConfirm <- []*UnConfirmedMessage{message}
-	//	}
-	//}
-	//producer.mutex.Unlock()
-	//producer.resetUnConfirmed()
+	producer.mutex.Lock()
+	if producer.publishConfirm != nil {
+		for _, message := range producer.unConfirmedMessages {
+			message.Confirmed = false
+			producer.publishConfirm <- []*UnConfirmedMessage{message}
+		}
+	}
+	producer.mutex.Unlock()
+	producer.resetUnConfirmed()
 
 	if producer.closeHandler != nil {
 		producer.closeHandler <- reason

pkg/stream/producer.go

@@ -69,11 +69,11 @@ func (producer *Producer) removeUnConfirmed(messageid int64) {
 	delete(producer.unConfirmedMessages, messageid)
 }
 
-//func (producer *Producer) resetUnConfirmed() {
-//	producer.mutex.Lock()
-//	defer producer.mutex.Unlock()
-//	producer.unConfirmedMessages = map[int64]*UnConfirmedMessage{}
-//}
+func (producer *Producer) resetUnConfirmed() {
+	producer.mutex.Lock()
+	defer producer.mutex.Unlock()
+	producer.unConfirmedMessages = map[int64]*UnConfirmedMessage{}
+}
 
 func (producer *Producer) lenUnConfirmed() int {
 	producer.mutex.Lock()