Merge pull request #99 from mreiferson/close_race_99

panic: runtime error: close of closed channel

consumer.go

@@ -123,6 +123,7 @@ type Consumer struct {
 	stopFlag        int32
 	connectedFlag   int32
 	stopHandler     sync.Once
+	exitHandler     sync.Once
 
 	// read from this channel to block until consumer is cleanly stopped
 	StopChan chan int
@@ -1079,9 +1080,11 @@ func (r *Consumer) shouldFailMessage(message *Message, handler interface{}) bool
 }
 
 func (r *Consumer) exit() {
-	close(r.exitChan)
-	r.wg.Wait()
-	close(r.StopChan)
+	r.exitHandler.Do(func() {
+		close(r.exitChan)
+		r.wg.Wait()
+		close(r.StopChan)
+	})
 }
 
 func (r *Consumer) log(lvl LogLevel, line string, args ...interface{}) {

producer.go

@@ -78,7 +78,6 @@ func NewProducer(addr string, config *Config) (*Producer, error) {
 		exitChan:        make(chan int),
 		responseChan:    make(chan []byte),
 		errorChan:       make(chan []byte),
-		closeChan:       make(chan int),
 	}
 	return p, nil
 }
@@ -223,7 +222,7 @@ func (w *Producer) connect() error {
 		atomic.StoreInt32(&w.state, StateInit)
 		return err
 	}
-
+	w.closeChan = make(chan int)
 	w.wg.Add(1)
 	go w.router()
 
@@ -323,4 +322,8 @@ func (w *Producer) onConnResponse(c *Conn, data []byte) { w.responseChan <- data
 func (w *Producer) onConnError(c *Conn, data []byte)    { w.errorChan <- data }
 func (w *Producer) onConnHeartbeat(c *Conn)             {}
 func (w *Producer) onConnIOError(c *Conn, err error)    { w.close() }
-func (w *Producer) onConnClose(c *Conn)                 { w.closeChan <- 1 }
+func (w *Producer) onConnClose(c *Conn) {
+	w.guard.Lock()
+	defer w.guard.Unlock()
+	close(w.closeChan)
+}