trying to stop the writer as soon as it's known that there is a communication issue

Author: scottf

src/main/java/io/nats/client/impl/NatsConnection.java

@@ -722,7 +722,7 @@ void handleCommunicationIssue(Exception io) {
             statusLock.unlock();
         }
 
-        executor.submit(() -> { writer.stop(); });
+        writer.preventWrite();
 
         processException(io);
         if (currentServer != null) {

src/main/java/io/nats/client/impl/NatsConnectionWriter.java

@@ -48,6 +48,7 @@ enum Mode {
     private Future<Boolean> stopped;
     private Future<DataPort> dataPortFuture;
     private DataPort dataPort;
+    private final AtomicBoolean allowedToWrite;
     private final AtomicBoolean running;
     private final AtomicReference<Mode> mode;
     private final ReentrantLock startStopLock;
@@ -66,6 +67,7 @@ enum Mode {
         writerLock = new ReentrantLock();
 
         this.running = new AtomicBoolean(false);
+        this.allowedToWrite = new AtomicBoolean(false);
         if (sourceWriter == null) {
             mode = new AtomicReference<>(Mode.Normal);
         }
@@ -100,6 +102,7 @@ void start(Future<DataPort> dataPortFuture) {
         this.startStopLock.lock();
         try {
             this.dataPortFuture = dataPortFuture;
+            this.allowedToWrite.set(true);
             this.running.set(true);
             this.normalOutgoing.resume();
             this.reconnectOutgoing.resume();
@@ -114,6 +117,7 @@ void start(Future<DataPort> dataPortFuture) {
     // method does.
     Future<Boolean> stop() {
         if (running.get()) {
+            allowedToWrite.set(false);
             running.set(false);
             startStopLock.lock();
             try {
@@ -128,6 +132,10 @@ Future<Boolean> stop() {
         return this.stopped;
     }
 
+    void preventWrite() {
+        allowedToWrite.set(false);
+    }
+
     boolean isRunning() {
         return running.get();
     }
@@ -219,7 +227,7 @@ public void run() {
         try {
             dataPort = this.dataPortFuture.get(); // Will wait for the future to complete
 
-            while (running.get() && !Thread.interrupted()) {
+            while (allowedToWrite.get() && !Thread.interrupted()) {
                 NatsMessage msg;
                 if (mode.get() == Mode.Normal) {
                     msg = this.normalOutgoing.accumulate(sendBufferLength.get(), Options.MAX_MESSAGES_IN_NETWORK_BUFFER, outgoingTimeout);
@@ -242,7 +250,8 @@ public void run() {
             // Exit
             Thread.currentThread().interrupt();
         } finally {
-            this.running.set(false);
+            allowedToWrite.set(false);
+            running.set(false);
         }
     }