Call pending callbacks of connection after run loop stopped (apache#239)

pulsar/internal/connection.go

@@ -147,6 +147,7 @@ type connection struct {
 	incomingRequestsCh chan *request
 	incomingCmdCh      chan *incomingCmd
 	closeCh            chan interface{}
+	runLoopStoppedCh   chan interface{}
 	writeRequestsCh    chan []byte
 
 	pendingReqs map[uint64]*request
@@ -176,6 +177,7 @@ func newConnection(logicalAddr *url.URL, physicalAddr *url.URL, tlsOptions *TLSO
 		auth:                 auth,
 
 		closeCh:            make(chan interface{}),
+		runLoopStoppedCh:   make(chan interface{}),
 		incomingRequestsCh: make(chan *request, 10),
 		incomingCmdCh:      make(chan *incomingCmd, 10),
 		writeRequestsCh:    make(chan []byte, 10),
@@ -312,6 +314,7 @@ func (c *connection) run() {
 	for {
 		select {
 		case <-c.closeCh:
+			close(c.runLoopStoppedCh)
 			c.Close()
 			return
 
@@ -656,18 +659,18 @@ 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 {
+	<-c.runLoopStoppedCh
+	for id, req := range c.pendingReqs {
 		req.callback(nil, errors.New("connection closed"))
+		delete(c.pendingReqs, id)
 	}
 
 	consumerHandlers := make(map[uint64]ConsumerHandler)