Fix close on closed channel

giautm · Mar 17, 2019 · b53ed77 · b53ed77
1 parent a5cb1ff
commit b53ed77
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 5 deletions.
diff --git a/.vscode/settings.json b/.vscode/settings.json
@@ -0,0 +1,5 @@
+{
+  "go.testEnvVars": {
+    "KAFKA_BROKERS": "localhost:9092",
+  }
+}
diff --git a/sarama.go b/sarama.go
@@ -2,6 +2,7 @@ package kafka
 
 import (
 	"context"
+	"sync"
 	"time"
 
 	"github.com/Shopify/sarama"
@@ -77,10 +78,11 @@ func (p *producer) Close() error {
 }
 
 type consumer struct {
-	ctx     context.Context
-	group   sarama.ConsumerGroup
-	topics  []string
-	started chan none
+	ctx         context.Context
+	group       sarama.ConsumerGroup
+	topics      []string
+	started     chan none
+	startedOnce sync.Once
 }
 
 func NewConsumer(ctx context.Context, group sarama.ConsumerGroup, topics []string) (KafkaConsumer, error) {
@@ -96,7 +98,9 @@ func (c *consumer) Receive(ctx context.Context, f HandlerFunc) error {
 	handler := &consumerGroupHandler{
 		handler: f,
 		started: func() {
-			close(c.started)
+			c.startedOnce.Do(func() {
+				close(c.started)
+			})
 		},
 	}