Clean callbacks of connection after run loop stopped (#239)

pulsar/internal/connection.go

@@ -126,6 +126,7 @@ type connection struct {
 	cond              *sync.Cond
 	state             connectionState
 	connectionTimeout time.Duration
+	closeOnce         sync.Once
 
 	logicalAddr  *url.URL
 	physicalAddr *url.URL
@@ -309,10 +310,19 @@ func (c *connection) run() {
 	go c.reader.readFromConnection()
 	go c.runPingCheck()
 
+	defer func() {
+		// all the accesses to the pendingReqs should be happened in this run loop thread,
+		// including the final cleanup, to avoid the issue https://github.com/apache/pulsar-client-go/issues/239
+		for id, req := range c.pendingReqs {
+			req.callback(nil, errors.New("connection closed"))
+			delete(c.pendingReqs, id)
+		}
+		c.Close()
+	}()
+
 	for {
 		select {
 		case <-c.closeCh:
-			c.Close()
 			return
 
 		case req := <-c.incomingRequestsCh:
@@ -361,7 +371,7 @@ func (c *connection) internalWriteData(data []byte) {
 	c.log.Debug("Write data: ", len(data))
 	if _, err := c.cnx.Write(data); err != nil {
 		c.log.WithError(err).Warn("Failed to write on connection")
-		c.Close()
+		c.TriggerClose()
 	}
 }
 
@@ -450,7 +460,7 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
 
 	default:
 		c.log.Errorf("Received invalid command type: %s", cmd.Type)
-		c.Close()
+		c.TriggerClose()
 	}
 }
 
@@ -568,7 +578,7 @@ func (c *connection) handleAuthChallenge(authChallenge *pb.CommandAuthChallenge)
 	authData, err := c.auth.GetData()
 	if err != nil {
 		c.log.WithError(err).Warn("Failed to load auth credentials")
-		c.Close()
+		c.TriggerClose()
 		return
 	}
 
@@ -630,18 +640,14 @@ func (c *connection) UnregisterListener(id uint64) {
 // Triggers the connection close by forcing the socket to close and
 // broadcasting the notification on the close channel
 func (c *connection) TriggerClose() {
-	cnx := c.cnx
-	if cnx != nil {
-		cnx.Close()
-	}
+	c.closeOnce.Do(func() {
+		cnx := c.cnx
+		if cnx != nil {
+			cnx.Close()
+		}
 
-	select {
-	case <-c.closeCh:
-		return
-	default:
 		close(c.closeCh)
-	}
-
+	})
 }
 
 func (c *connection) Close() {
@@ -656,20 +662,14 @@ func (c *connection) Close() {
 
 	c.log.Info("Connection closed")
 	c.state = connectionClosed
-	if c.cnx != nil {
-		c.cnx.Close()
-	}
+	c.TriggerClose()
 	c.pingTicker.Stop()
 	c.pingCheckTicker.Stop()
 
 	for _, listener := range c.listeners {
 		listener.ConnectionClosed()
 	}
 
-	for _, req := range c.pendingReqs {
-		req.callback(nil, errors.New("connection closed"))
-	}
-
 	consumerHandlers := make(map[uint64]ConsumerHandler)
 	c.consumerHandlersLock.RLock()
 	for id, handler := range c.consumerHandlers {