While handleCommunicationIssue, closeSocket should use reconnect impl otherwise tryingToConnect will guard (#1206)

* While `handleCommunicationIssue`, `closeSocket` should use reconnect impl otherwise `tryingToConnect` will guard

Signed-off-by: Maurice van Veen <[email protected]>

* Fix NPE on dataPortFuture with low connection timeout

Signed-off-by: Maurice van Veen <[email protected]>

* During forceReconnect, ensure reader/writer is stopped before continuing

---------

Signed-off-by: Maurice van Veen <[email protected]>
Co-authored-by: scottf <[email protected]>

Author: MauriceVanVeen

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

@@ -326,12 +326,12 @@ void forceReconnectImpl(ForceReconnectOptions options) throws InterruptedExcepti
 
             // stop i/o
             try {
-                this.reader.stop(false).get(10, TimeUnit.SECONDS);
+                this.reader.stop(false).get(100, TimeUnit.MILLISECONDS);
             } catch (Exception ex) {
                 processException(ex);
             }
             try {
-                this.writer.stop().get(10, TimeUnit.SECONDS);
+                this.writer.stop().get(100, TimeUnit.MILLISECONDS);
             } catch (Exception ex) {
                 processException(ex);
             }
@@ -765,7 +765,7 @@ void closeSocket(boolean tryReconnectIfConnected, boolean forceClose) throws Int
             if (isClosing()) { // isClosing() means we are in the close method or were asked to be
                 close();
             } else if (wasConnected && tryReconnectIfConnected) {
-                reconnect();
+                reconnectImpl(); // call the impl here otherwise the tryingToConnect guard will block the behavior
             }
         } finally {
             closeSocketLock.unlock();
@@ -880,7 +880,11 @@ void closeSocketImpl(boolean forceClose) {
             //
         }
 
-        this.dataPortFuture.cancel(true);
+        // Close and reset the current data port and future
+        if (dataPortFuture != null) {
+            dataPortFuture.cancel(true);
+            dataPortFuture = null;
+        }
 
         // Close the current socket and cancel anyone waiting for it
         try {

src/test/java/io/nats/client/ConnectTests.java

@@ -617,4 +617,17 @@ private static void validateRunInJsCluster(Connection nc1, Connection nc2, Conne
         assertTrue(urls3.contains(port2));
         assertTrue(urls3.contains(port3));
     }
+
+    // https://github.com/nats-io/nats.java/issues/1201
+    @Test
+    void testLowConnectionTimeoutResultsInIOException() {
+        Options options = Options.builder()
+                .connectionTimeout(Duration.ZERO)
+                .build();
+
+        assertThrows(IOException.class, () -> {
+            Connection nc = Nats.connect(options);
+            nc.close();
+        });
+    }
 }