Null-proof channel state retrieval in AbstractMetricsCollector

A channel message may have been closed by the time a metrics method is
called (e.g. when cancelling a consumer the metrics collector is called
after the RPC is completed), so the channel state may no longer be
there.

This commit null-proofs methods rely on the channel state retrieval.

Fixes #1592

Author: acogoluegnes

src/main/java/com/rabbitmq/client/impl/AbstractMetricsCollector.java

@@ -115,9 +115,12 @@ public void basicPublish(Channel channel, long deliveryTag) {
         try {
             if (deliveryTag != 0) {
                 ChannelState channelState = channelState(channel);
+                if (channelState == null) {
+                    return;
+                }
                 channelState.lock.lock();
                 try {
-                    channelState(channel).unconfirmedMessageDeliveryTags.add(deliveryTag);
+                    channelState.unconfirmedMessageDeliveryTags.add(deliveryTag);
                 } finally {
                     channelState.lock.unlock();
                 }
@@ -169,9 +172,12 @@ public void basicConsume(Channel channel, String consumerTag, boolean autoAck) {
         try {
             if(!autoAck) {
                 ChannelState channelState = channelState(channel);
+                if (channelState == null) {
+                    return;
+                }
                 channelState.lock.lock();
                 try {
-                    channelState(channel).consumersWithManualAck.add(consumerTag);
+                    channelState.consumersWithManualAck.add(consumerTag);
                 } finally {
                     channelState.lock.unlock();
                 }
@@ -185,9 +191,12 @@ public void basicConsume(Channel channel, String consumerTag, boolean autoAck) {
     public void basicCancel(Channel channel, String consumerTag) {
         try {
             ChannelState channelState = channelState(channel);
+            if (channelState == null) {
+                return;
+            }
             channelState.lock.lock();
             try {
-                channelState(channel).consumersWithManualAck.remove(consumerTag);
+                channelState.consumersWithManualAck.remove(consumerTag);
             } finally {
                 channelState.lock.unlock();
             }
@@ -202,9 +211,12 @@ public void consumedMessage(Channel channel, long deliveryTag, boolean autoAck)
             markConsumedMessage();
             if(!autoAck) {
                 ChannelState channelState = channelState(channel);
+                if (channelState == null) {
+                    return;
+                }
                 channelState.lock.lock();
                 try {
-                    channelState(channel).unackedMessageDeliveryTags.add(deliveryTag);
+                    channelState.unackedMessageDeliveryTags.add(deliveryTag);
                 } finally {
                     channelState.lock.unlock();
                 }
@@ -219,6 +231,9 @@ public void consumedMessage(Channel channel, long deliveryTag, String consumerTa
         try {
             markConsumedMessage();
             ChannelState channelState = channelState(channel);
+            if (channelState == null) {
+                return;
+            }
             channelState.lock.lock();
             try {
                 if(channelState.consumersWithManualAck.contains(consumerTag)) {