Always check connection close channell, before attempting to put requests (#521)
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index 1c264c6..9cc1238 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -549,10 +549,15 @@
if c.getState() == connectionClosed {
callback(req, ErrConnectionClosed)
} else {
- c.incomingRequestsCh <- &request{
+ select {
+ case <-c.closeCh:
+ callback(req, ErrConnectionClosed)
+
+ case c.incomingRequestsCh <- &request{
id: &requestID,
cmd: req,
callback: callback,
+ }:
}
}
}
@@ -562,12 +567,17 @@
return ErrConnectionClosed
}
- c.incomingRequestsCh <- &request{
+ select {
+ case <-c.closeCh:
+ return ErrConnectionClosed
+
+ case c.incomingRequestsCh <- &request{
id: nil,
cmd: req,
callback: nil,
+ }:
+ return nil
}
- return nil
}
func (c *connection) internalSendRequest(req *request) {
