Fix issue when channel/connection is closed before NotifyClose

This fixes an issue where the connection or channel would be closed
before NotifyClose is setup. That can cause the monitorAndWait to block
even when a connection or channel is actually closed. This happens
because the consumers or publishers can close the channel due to an
error inside their respective goroutines.

This fix also avoids starting and leaking a bunch of goroutines inside
monitorAndWait.

Author: akarl

client.go

@@ -360,6 +360,14 @@ func (c *Client) runOnce() error {
 		onStarted(inputConn, outputConn, inputCh, outputCh)
 	}
 
+	// Start listening on NotifyClose before we properly begin so that we avoid
+	// race when the consumer or publisher starts work before we call
+	// monitorAndWait. All have a buffer of 1 as recommended by amqp-go.
+	notifyInputConnClose := inputConn.NotifyClose(make(chan *amqp.Error, 1))
+	notifyOutputConnClose := outputConn.NotifyClose(make(chan *amqp.Error, 1))
+	notifyInputChClose := inputCh.NotifyClose(make(chan *amqp.Error, 1))
+	notifyOutputChClose := outputCh.NotifyClose(make(chan *amqp.Error, 1))
+
 	err = c.runRepliesConsumer(inputCh)
 	if err != nil {
 		return err
@@ -386,10 +394,10 @@ func (c *Client) runOnce() error {
 	_, err = monitorAndWait(
 		make(chan struct{}),
 		c.stopChan,
-		inputConn.NotifyClose(make(chan *amqp.Error)),
-		outputConn.NotifyClose(make(chan *amqp.Error)),
-		inputCh.NotifyClose(make(chan *amqp.Error)),
-		outputCh.NotifyClose(make(chan *amqp.Error)),
+		notifyInputConnClose,
+		notifyOutputConnClose,
+		notifyInputChClose,
+		notifyOutputChClose,
 	)
 	if err != nil {
 		return err

connection.go

@@ -19,29 +19,39 @@ var ErrUnexpectedConnClosed = errors.New("unexpected connection close without sp
 // NotifyPublish etc.
 type OnStartedFunc func(inputConn, outputConn *amqp.Connection, inputChannel, outputChannel *amqp.Channel)
 
-func monitorAndWait(restartChan, stopChan chan struct{}, amqpErrs ...chan *amqp.Error) (bool, error) {
-	result := make(chan error, len(amqpErrs))
-
-	// Setup monitoring for connections and channels, can be several connections and several channels.
-	// The first one closed will yield the error.
-	for _, errCh := range amqpErrs {
-		go func(c chan *amqp.Error) {
-			err, ok := <-c
-			if !ok {
-				result <- ErrUnexpectedConnClosed
-				return
-			}
-			result <- err
-		}(errCh)
-	}
-
+func monitorAndWait(
+	restartChan,
+	stopChan chan struct{},
+	inputConnClose,
+	outputConnClose,
+	inputChClose,
+	outputChClose chan *amqp.Error,
+) (bool, error) {
 	select {
-	case err := <-result:
-		return true, err
 	case <-restartChan:
 		return true, nil
 	case <-stopChan:
 		return false, nil
+	case err, ok := <-inputConnClose:
+		if !ok {
+			return false, ErrUnexpectedConnClosed
+		}
+		return false, err
+	case err, ok := <-outputConnClose:
+		if !ok {
+			return false, ErrUnexpectedConnClosed
+		}
+		return false, err
+	case err, ok := <-inputChClose:
+		if !ok {
+			return false, ErrUnexpectedConnClosed
+		}
+		return false, err
+	case err, ok := <-outputChClose:
+		if !ok {
+			return false, ErrUnexpectedConnClosed
+		}
+		return false, err
 	}
 }
 

server.go

@@ -274,6 +274,14 @@ func (s *Server) listenAndServe() (bool, error) {
 	// Notify everyone that the server has started.
 	s.notifyStarted(inputConn, outputConn, inputCh, outputCh)
 
+	// Start listening on NotifyClose before we properly begin so that we avoid
+	// race when the consumer or publisher starts work before we call
+	// monitorAndWait. All have a buffer of 1 as recommended by amqp-go.
+	notifyInputConnClose := inputConn.NotifyClose(make(chan *amqp.Error, 1))
+	notifyOutputConnClose := outputConn.NotifyClose(make(chan *amqp.Error, 1))
+	notifyInputChClose := inputCh.NotifyClose(make(chan *amqp.Error, 1))
+	notifyOutputChClose := outputCh.NotifyClose(make(chan *amqp.Error, 1))
+
 	// Create any exchanges that must be declared on startup.
 	err = createExchanges(inputCh, s.exchanges)
 	if err != nil {
@@ -302,10 +310,10 @@ func (s *Server) listenAndServe() (bool, error) {
 	shouldRestart, err := monitorAndWait(
 		s.restartChan,
 		s.stopChan,
-		inputConn.NotifyClose(make(chan *amqp.Error)),
-		outputConn.NotifyClose(make(chan *amqp.Error)),
-		inputCh.NotifyClose(make(chan *amqp.Error)),
-		outputCh.NotifyClose(make(chan *amqp.Error)),
+		notifyInputConnClose,
+		notifyOutputConnClose,
+		notifyInputChClose,
+		notifyOutputChClose,
 	)
 	if err != nil {
 		return shouldRestart, err