fix data race of Consumer.closeHandler (#205)

* fix data race of Consumer.closeHandler

* use defer

Author: kensou97

pkg/stream/consumer.go

@@ -66,16 +66,14 @@ func (consumer *Consumer) setCurrentOffset(offset int64) {
 
 func (consumer *Consumer) GetOffset() int64 {
 	consumer.mutex.Lock()
-	res := consumer.currentOffset
-	consumer.mutex.Unlock()
-	return res
+	defer consumer.mutex.Unlock()
+	return consumer.currentOffset
 }
 
 func (consumer *Consumer) GetLastStoredOffset() int64 {
 	consumer.mutex.Lock()
-	res := consumer.lastStoredOffset
-	consumer.mutex.Unlock()
-	return res
+	defer consumer.mutex.Unlock()
+	return consumer.lastStoredOffset
 }
 
 func (consumer *Consumer) updateLastStoredOffset() bool {
@@ -88,6 +86,12 @@ func (consumer *Consumer) updateLastStoredOffset() bool {
 	return false
 }
 
+func (consumer *Consumer) GetCloseHandler() chan Event {
+	consumer.mutex.Lock()
+	defer consumer.mutex.Unlock()
+	return consumer.closeHandler
+}
+
 func (consumer *Consumer) NotifyClose() ChannelClose {
 	consumer.mutex.Lock()
 	defer consumer.mutex.Unlock()

pkg/stream/coordinator.go

@@ -82,9 +82,9 @@ func (coordinator *Coordinator) RemoveConsumerById(id interface{}, reason Event)
 	reason.StreamName = consumer.GetStreamName()
 	reason.Name = consumer.GetName()
 
-	if consumer.closeHandler != nil {
-		consumer.closeHandler <- reason
-		close(consumer.closeHandler)
+	if closeHandler := consumer.GetCloseHandler(); closeHandler != nil {
+		closeHandler <- reason
+		close(closeHandler)
 	}
 
 	return coordinator.removeById(id, coordinator.consumers)