fix: panic on close with full internal queue (#412)

Author: hiimjako

pkg/stream/consumer.go

@@ -342,8 +342,14 @@ func (consumer *Consumer) close(reason Event) {
 	}
 
 	if consumer.response.data != nil {
+		// drain the queue to avoid race condition
+		for len(consumer.chunkForConsumer) > 0 {
+			select {
+			case <-consumer.chunkForConsumer:
+			default:
+			}
+		}
 		close(consumer.chunkForConsumer)
-
 		close(consumer.response.data)
 		consumer.response.data = nil
 	}

pkg/stream/consumer_test.go

@@ -472,6 +472,35 @@ var _ = Describe("Streaming Consumers", func() {
 
 	})
 
+	It("Not panics on close when the internal queue is full", func() {
+		producer, err := env.NewProducer(streamName, nil)
+		Expect(err).NotTo(HaveOccurred())
+		err = producer.BatchSend(CreateArrayMessagesForTesting(10_000))
+		Expect(err).NotTo(HaveOccurred())
+
+		defer func(producer *Producer) {
+			err := producer.Close()
+			Expect(err).NotTo(HaveOccurred())
+		}(producer)
+
+		const credits = 2
+		consumer, err := env.NewConsumer(streamName,
+			func(_ ConsumerContext, _ *amqp.Message) {
+				// it usually happens with slow consumer that fulfill the internal queue
+				time.Sleep(time.Hour)
+			}, NewConsumerOptions().
+				SetInitialCredits(credits).
+				SetOffset(OffsetSpecification{}.First()).
+				SetConsumerName("consumer_test"))
+		Expect(err).NotTo(HaveOccurred())
+
+		// waiting the internal queue to fulfill
+		time.Sleep(time.Second)
+		Expect(len(consumer.chunkForConsumer)).To(Equal(credits))
+
+		Expect(consumer.Close()).NotTo(HaveOccurred())
+	})
+
 	It("message Properties", func() {
 		producer, err := env.NewProducer(streamName, nil)
 

pkg/stream/server_frame.go

@@ -404,13 +404,13 @@ func (c *Client) handleDeliver(r *bufio.Reader) {
 			}
 		}
 	}
-	// request a credit for the next chunk
-	c.credit(subscriptionId, 1)
 
 	// dispatch the messages with offset to the consumer
 	chunk.offsetMessages = batchConsumingMessages
 	if consumer.getStatus() == open {
 		consumer.chunkForConsumer <- chunk
+		// request a credit for the next chunk
+		c.credit(subscriptionId, 1)
 	} else {
 		logs.LogDebug("The consumer %s for the stream %s is closed during the chunk dispatching. "+
 			"Messages won't dispatched", consumer.GetName(), consumer.GetStreamName())