Trying to ensure no blocking close on force reconnect from socket write timeout

Author: scottf

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

@@ -293,18 +293,21 @@ void forceReconnectImpl() throws IOException, InterruptedException {
                 dataPortFuture.cancel(true);
                 dataPortFuture = null;
             }
+
+            // close the data port as a task so as not to block reconnect
             if (dataPort != null) {
-                try {
-                    dataPort.close();
-                }
-                catch (IOException ignore) {}
-                finally {
-                    dataPort = null;
-                }
+                final DataPort closeMe = dataPort;
+                dataPort = null;
+                executor.submit(() -> {
+                    try {
+                        closeMe.close();
+                    }
+                    catch (IOException ignore) {}
+                });
             }
 
             // stop i/o
-            reader.stop();
+            reader.stop(false);
             writer.stop();
 
             // new reader/writer

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

@@ -94,13 +94,17 @@ void start(Future<DataPort> dataPortFuture) {
         this.stopped = connection.getExecutor().submit(this, Boolean.TRUE);
     }
 
+    Future<Boolean> stop() {
+        return stop(true);
+    }
+
     // May be called several times on an error.
     // Returns a future that is completed when the thread completes, not when this
     // method does.
-    Future<Boolean> stop() {
+    Future<Boolean> stop(boolean shutdownDataPort) {
         if (running.get()) {
             running.set(false);
-            if (dataPort != null) {
+            if (shutdownDataPort && dataPort != null) {
                 try {
                     dataPort.shutdownInput();
                 }

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

@@ -38,12 +38,15 @@ public void run() {
             if (System.nanoTime() > writeMustBeDoneBy) {
                 writeWatcherTimer.cancel(); // we don't need to repeat this
                 connection.executeCallback((c, el) -> el.socketWriteTimeout(c));
-                connection.getExecutor().submit(() -> {
-                    try {
-                        connection.forceReconnect();
-                    }
-                    catch (IOException | InterruptedException ignore) {}
-                });
+                try {
+                    connection.forceReconnect();
+                }
+                catch (IOException e) {
+                    // retry maybe? forceReconnect
+                }
+                catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
             }
         }
     }