Check if the producer is closed when send

pkg/stream/constants.go

@@ -81,9 +81,8 @@ const (
 	responseCodeNoOffset                      = uint16(19)
 
 	/// responses out of protocol
-	closeChannel         = uint16(60)
-	connectionCloseError = uint16(61)
-	timeoutError         = uint16(62)
+	timeoutError = uint16(62)
+	entityClosed = uint16(63)
 
 	///
 	defaultSocketCallTimeout = 10 * time.Second
@@ -194,6 +193,9 @@ func lookErrorCode(errorCode uint16) error {
 		return AuthenticationFailureLoopbackError
 	case timeoutError:
 		return ConfirmationTimoutError
+	case entityClosed:
+		return AlreadyClosed
+
 	default:
 		{
 			logs.LogWarn("Error not handled %d", errorCode)

pkg/stream/producer.go

@@ -362,10 +362,25 @@ func (producer *Producer) fromMessageToMessageSequence(streamMessage message.Str
 // and the messages in the buffer. The aggregation is up to the client.
 // returns an error if the message could not be sent for marshal problems or if the buffer is too large
 func (producer *Producer) Send(streamMessage message.StreamMessage) error {
+
 	messageSeq, err := producer.fromMessageToMessageSequence(streamMessage)
 	if err != nil {
 		return err
 	}
+	if producer.getStatus() == closed {
+		producer.sendConfirmationStatus([]*ConfirmationStatus{
+			{
+				inserted:     time.Now(),
+				message:      streamMessage,
+				producerID:   producer.GetID(),
+				publishingId: messageSeq.publishingId,
+				confirmed:    false,
+				err:          AlreadyClosed,
+			},
+		})
+		return fmt.Errorf("producer id: %d closed", producer.id)
+	}
+
 	producer.unConfirmed.addFromSequence(messageSeq, &streamMessage, producer.GetID())
 
 	if len(messageSeq.messageBytes) > defaultMaxFrameSize {
@@ -377,7 +392,7 @@ func (producer *Producer) Send(streamMessage message.StreamMessage) error {
 	// se the processPendingSequencesQueue function
 	err = producer.pendingSequencesQueue.Enqueue(messageSeq)
 	if err != nil {
-		return fmt.Errorf("error during enqueue message: %s. Message will be in timed. Producer id: %d ", err, producer.id)
+		return fmt.Errorf("error during enqueue message: %s pending queue closed. Producer id: %d ", err, producer.id)
 	}
 	return nil
 }
@@ -403,6 +418,14 @@ func (producer *Producer) BatchSend(batchMessages []message.StreamMessage) error
 	}
 	//
 
+	if producer.getStatus() == closed {
+		for _, msg := range messagesSequence {
+			m := producer.unConfirmed.extractWithError(msg.publishingId, entityClosed)
+			producer.sendConfirmationStatus([]*ConfirmationStatus{m})
+		}
+		return fmt.Errorf("producer id: %d closed", producer.id)
+	}
+
 	if totalBufferToSend+initBufferPublishSize > maxFrame {
 		// if the totalBufferToSend is greater than the requestedMaxFrameSize
 		// all the messages are unconfirmed

pkg/stream/producer_test.go

@@ -806,12 +806,21 @@ var _ = Describe("Streaming Producers", func() {
 	})
 
 	It("Can't send message if the producer is closed", func() {
-
 		producer, err := testEnvironment.NewProducer(testProducerStream, nil)
 		Expect(err).NotTo(HaveOccurred())
 		Expect(producer.Close()).NotTo(HaveOccurred())
 		err = producer.Send(amqp.NewMessage(make([]byte, 50)))
 		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("producer id: 0 closed"))
+	})
+
+	It("Can't bach send message if the producer is closed", func() {
+		producer, err := testEnvironment.NewProducer(testProducerStream, nil)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(producer.Close()).NotTo(HaveOccurred())
+		err = producer.BatchSend([]message.StreamMessage{amqp.NewMessage(make([]byte, 50))})
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("producer id: 0 closed"))
 	})
 
 })