Add support for Sink.Close() being called multiple times

integration_test.go

@@ -219,6 +219,13 @@ func (s *sinkTestSuite) TestSend() {
 	s.Equal(expectedMessages, receivedMessages)
 }
 
+func (s *sinkTestSuite) TestCloseTwice() {
+	err := s.sink.Close()
+	s.Nil(err)
+	err = s.sink.Close()
+	s.Nil(err)
+}
+
 func (s *sourceTestSuite) produce(values []string) {
 	for _, m := range values {
 		s.prod.Produce(&kafka.Message{

sink.go

@@ -23,8 +23,8 @@ var (
 // Sink encapsulates a kafka producer for Sending Msgs
 type Sink struct {
 	prod     *kafka.Producer
-	quitChan chan int
-	doneChan chan int
+	quitChan chan struct{}
+	doneChan chan struct{}
 	evtChan  chan frizzle.Event
 }
 
@@ -48,8 +48,8 @@ func InitSink(config *viper.Viper) (*Sink, error) {
 
 	s := &Sink{
 		prod:     p,
-		quitChan: make(chan int),
-		doneChan: make(chan int),
+		quitChan: make(chan struct{}),
+		doneChan: make(chan struct{}),
 		evtChan:  make(chan frizzle.Event),
 	}
 
@@ -62,11 +62,11 @@ func InitSink(config *viper.Viper) (*Sink, error) {
 // message delivery is successful, any errors from broker, etc
 func (s *Sink) deliveryReports() {
 	defer close(s.doneChan)
-	run := true
-	for run == true {
+	for {
 		select {
 		case <-s.quitChan:
-			run = false
+			s.quitChan = nil
+			return
 		case e := <-s.prod.Events():
 			switch ev := e.(type) {
 			case *kafka.Message:
@@ -104,18 +104,21 @@ func (s *Sink) Send(m frizzle.Msg, topic string) error {
 // Close the Sink after flushing any Msgs not fully sent
 func (s *Sink) Close() error {
 	// Flush any messages still pending send
-	fmt.Println("flush producer")
 	if remaining := s.prod.Flush(flushTimeoutMS); remaining > 0 {
 		return fmt.Errorf("there are still %d messages which have not been delivered after %d milliseconds", remaining, flushTimeoutMS)
 	}
-	// tell deliveryReports() goroutine to finish
-	fmt.Println("close deliveryReports() loop")
-	s.quitChan <- 1
+
+	// check if already closed, return if so
+	if s.quitChan == nil {
+		return nil
+	}
+
+	// tell deliveryReports() goroutine to finish if running
+	close(s.quitChan)
 	// wait for it to finish
 	<-s.doneChan
 	// stop event chan
 	close(s.evtChan)
-	fmt.Println("close producer")
 	s.prod.Close()
 	return nil
 }

source.go

@@ -182,7 +182,6 @@ func (s *Source) Ping() error {
 // unAcked Msgs.
 func (s *Source) Close() error {
 	// confirm that consume() goroutine finished
-	fmt.Println("begin source close")
 	select {
 	case <-s.doneChan:
 	case <-time.After(stopCloseTimeout):
@@ -191,9 +190,7 @@ func (s *Source) Close() error {
 	if s.unAcked.Count() > 0 {
 		return frizzle.ErrUnackedMsgsRemain
 	}
-	fmt.Println("close channels")
 	close(s.msgChan)
 	close(s.evtChan)
-	fmt.Println("close consumer")
 	return s.cons.Close()
 }