Use dedicated flag to dispatch frame on connection closing

References #585

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/Client.java

@@ -158,6 +158,7 @@ public class Client implements AutoCloseable {
   final ExecutorService dispatchingExecutorService;
   final TuneState tuneState;
   final AtomicBoolean closing = new AtomicBoolean(false);
+  final AtomicBoolean shuttingDownDispatching = new AtomicBoolean(false);
   final ChunkChecksum chunkChecksum;
   final MetricsCollector metricsCollector;
   final CompressionCodecFactory compressionCodecFactory;
@@ -354,6 +355,7 @@ public void initChannel(SocketChannel ch) {
             () -> {
               if (dispatchingExecutorServiceFactory == null) {
                 List<Runnable> outstandingTasks = this.dispatchingExecutorService.shutdownNow();
+                this.shuttingDownDispatching.set(true);
                 for (Runnable outstandingTask : outstandingTasks) {
                   try {
                     outstandingTask.run();
@@ -2778,7 +2780,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
         }
       } else {
         FrameHandler frameHandler = ServerFrameHandler.lookup(commandId, version, m);
-        task = new FrameHandlerTask(frameHandler, Client.this, frameSize, ctx, m, closing);
+        task =
+            new FrameHandlerTask(
+                frameHandler, Client.this, frameSize, ctx, m, shuttingDownDispatching);
       }
 
       if (task != null) {
@@ -2854,26 +2858,26 @@ private static class FrameHandlerTask implements Runnable {
     private final int frameSize;
     private final ChannelHandlerContext ctx;
     private final ByteBuf message;
-    private final AtomicBoolean closing;
+    private final AtomicBoolean shouldRelease;
 
     private FrameHandlerTask(
         FrameHandler frameHandler,
         Client client,
         int frameSize,
         ChannelHandlerContext ctx,
         ByteBuf message,
-        AtomicBoolean closing) {
+        AtomicBoolean shouldRelease) {
       this.frameHandler = frameHandler;
       this.client = client;
       this.frameSize = frameSize;
       this.ctx = ctx;
       this.message = message;
-      this.closing = closing;
+      this.shouldRelease = shouldRelease;
     }
 
     @Override
     public void run() {
-      if (this.closing.get()) {
+      if (this.shouldRelease.get()) {
         try {
           this.message.release();
         } catch (Exception e) {

src/test/java/com/rabbitmq/stream/impl/AuthenticationTest.java

@@ -105,9 +105,7 @@ public byte[] handleChallenge(
                       }));
     } catch (StreamException e) {
       // there can be a timeout because the connection gets closed before returning the error
-      assertThat(e.getMessage())
-          .containsAnyOf(
-              String.valueOf(Constants.RESPONSE_CODE_SASL_ERROR), "Could not get response in");
+      assertThat(e).hasMessageContaining(String.valueOf(Constants.RESPONSE_CODE_SASL_ERROR));
     }
   }