handle closing

Author: Gsantomaggio

README.md

@@ -8,7 +8,7 @@ Experimental client for [RabbitMQ Stream Queues](https://github.com/rabbitmq/rab
 ---
 
 ```
-go get -u github.com/rabbitmq/rabbitmq-stream-go-client@v0.9-alpha
+go get -u github.com/rabbitmq/rabbitmq-stream-go-client@v0.10-alpha
 ```
 
 ### Getting started

VERSION

@@ -1 +1 @@
-0.9-alpha
+0.10-alpha

pkg/ha/ha_publisher.go

@@ -73,14 +73,6 @@ func NewHAProducer(env *stream.Environment, streamName string, producerOptions *
 
 func (p *ReliableProducer) newProducer() error {
 
-	//if p.producer != nil && len(p.producer.GetUnConfirmed()) > 0 {
-	//	for _, msg := range p.producer.GetUnConfirmed() {
-	//		msg.Confirmed = false
-	//		p.channelPublishConfirm <- []*stream.UnConfirmedMessage{msg}
-	//	}
-	//
-	//}
-
 	producer, err := p.env.NewProducer(p.streamName, p.producerOptions)
 	if err != nil {
 		return err

pkg/stream/constants.go

@@ -11,8 +11,8 @@ import (
 // for example the producer status
 
 const (
-	running = iota
-	closed  = iota
+	open   = iota
+	closed = iota
 )
 
 const initBufferPublishSize = 2 + 2 + 1 + 4
@@ -79,11 +79,13 @@ const (
 	defaultSocketBuffer = 4096
 
 	//
-	ClientVersion = "0.9-alpha"
+	ClientVersion = "0.10-alpha"
 
 	StreamTcpPort = "5552"
 )
 
+var AlreadyClosed = errors.New("Already Closed")
+
 var PreconditionFailed = errors.New("Precondition Failed")
 var AuthenticationFailure = errors.New("Authentication Failure")
 var StreamDoesNotExist = errors.New("Stream Does Not Exist")

pkg/stream/consumer.go

@@ -18,7 +18,21 @@ type Consumer struct {
 	// different form ConsumerOptions.offset. ConsumerOptions.offset is just the configuration
 	// and won't change. currentOffset is the status of the offset
 	currentOffset int64
-	CloseHandler  chan Event
+	closeHandler  chan Event
+
+	status int
+}
+
+func (consumer *Consumer) setStatus(status int) {
+	consumer.mutex.Lock()
+	defer consumer.mutex.Unlock()
+	consumer.status = status
+}
+
+func (consumer *Consumer) getStatus() int {
+	consumer.mutex.Lock()
+	defer consumer.mutex.Unlock()
+	return consumer.status
 }
 
 func (consumer *Consumer) GetStreamName() string {
@@ -50,7 +64,7 @@ func (consumer *Consumer) GetOffset() int64 {
 
 func (consumer *Consumer) NotifyClose() ChannelClose {
 	ch := make(chan Event, 1)
-	consumer.CloseHandler = ch
+	consumer.closeHandler = ch
 	return ch
 }
 
@@ -105,6 +119,10 @@ func (c *Client) credit(subscriptionId byte, credit int16) {
 }
 
 func (consumer *Consumer) Close() error {
+	if consumer.getStatus() == closed {
+		return AlreadyClosed
+	}
+	consumer.setStatus(closed)
 	_, errGet := consumer.options.client.coordinator.GetConsumerById(consumer.ID)
 	if errGet != nil {
 		return nil

pkg/stream/consumer_test.go

@@ -239,4 +239,31 @@ var _ = Describe("Streaming Consumers", func() {
 		Expect(err).NotTo(HaveOccurred())
 	})
 
+	It("Check already closed", func() {
+		producer, err := env.NewProducer(streamName, nil)
+		Expect(err).NotTo(HaveOccurred())
+		err = producer.BatchSend(CreateArrayMessagesForTesting(500)) // batch send
+		Expect(err).NotTo(HaveOccurred())
+		defer func(producer *Producer) {
+			err := producer.Close()
+			Expect(err).NotTo(HaveOccurred())
+		}(producer)
+
+		var messagesCount int32 = 0
+		consumer, err := env.NewConsumer(streamName,
+			func(consumerContext ConsumerContext, message *amqp.Message) {
+				if atomic.AddInt32(&messagesCount, 1) >= 250 {
+					err := consumerContext.Consumer.Close()
+					if err != nil {
+						return
+					}
+				}
+			}, NewConsumerOptions().SetOffset(OffsetSpecification{}.First()).SetConsumerName("consumer_test"))
+		Expect(err).NotTo(HaveOccurred())
+		time.Sleep(500 * time.Millisecond)
+		err = consumer.Close()
+		Expect(err).To(Equal(AlreadyClosed))
+
+	})
+
 })

pkg/stream/coordinator.go

@@ -56,7 +56,7 @@ func (coordinator *Coordinator) NewProducer(
 		options:             parameters,
 		mutex:               &sync.Mutex{},
 		unConfirmedMessages: map[int64]*UnConfirmedMessage{},
-		status:              running,
+		status:              open,
 		messageSequenceCh:   make(chan messageSequence, size),
 		pendingMessages: pendingMessagesSequence{
 			messages: make([]messageSequence, 0),
@@ -74,8 +74,8 @@ func (coordinator *Coordinator) RemoveConsumerById(id interface{}, reason Event)
 	reason.StreamName = consumer.GetStreamName()
 	reason.Name = consumer.GetName()
 
-	if consumer.CloseHandler != nil {
-		consumer.CloseHandler <- reason
+	if consumer.closeHandler != nil {
+		consumer.closeHandler <- reason
 	}
 	return coordinator.removeById(id, coordinator.consumers)
 }
@@ -121,7 +121,7 @@ func (coordinator *Coordinator) ProducersCount() int {
 func newResponse(commandDescription string) *Response {
 	res := &Response{}
 	res.commandDescription = commandDescription
-	res.code = make(chan Code)
+	res.code = make(chan Code, 1)
 	res.data = make(chan interface{})
 	res.messages = make(chan []*amqp.Message, 100)
 	return res
@@ -188,7 +188,9 @@ func (coordinator *Coordinator) NewConsumer(messagesHandler MessagesHandler,
 	defer coordinator.mutex.Unlock()
 	var lastId, _ = coordinator.getNextConsumerItem()
 	var item = &Consumer{ID: lastId, options: parameters,
-		response: newResponse(lookUpCommand(commandSubscribe)), mutex: &sync.Mutex{},
+		response:        newResponse(lookUpCommand(commandSubscribe)),
+		status:          open,
+		mutex:           &sync.Mutex{},
 		MessagesHandler: messagesHandler,
 	}
 

pkg/stream/dialer_posix.go

@@ -30,4 +30,4 @@ func init() {
 
 		})
 	}
-}
+}

pkg/stream/dialer_windows.go

@@ -30,4 +30,4 @@ func init() {
 
 		})
 	}
-}
+}

pkg/stream/producer.go

@@ -302,9 +302,12 @@ func (producer *Producer) FlushUnConfirmedMessages() {
 }
 
 func (producer *Producer) Close() error {
+	if producer.getStatus() == closed {
+		return AlreadyClosed
+	}
 	producer.setStatus(closed)
 	if !producer.options.client.socket.isOpen() {
-		return fmt.Errorf("connection already closed")
+		return fmt.Errorf("tcp connection is closed")
 	}
 
 	err := producer.options.client.deletePublisher(producer.ID)