handleCommunicationIssue uses forceClose

Author: scottf

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

@@ -628,7 +628,9 @@ public void run() {
         } catch (Exception exp) {
             processException(exp);
             try {
-                this.closeSocket(false);
+                // allow force reconnect since this is pretty exceptional,
+                // a connection failure while trying to connect
+                this.closeSocket(false, true);
             } catch (InterruptedException e) {
                 processException(e);
             }
@@ -691,7 +693,9 @@ void handleCommunicationIssue(Exception io) {
         // waiting on read/write threads
         executor.submit(() -> {
             try {
-                this.closeSocket(true);
+                // any issue that brings us here is pretty serious
+                // so we are comfortable forcing the close
+                this.closeSocket(true, true);
             } catch (InterruptedException e) {
                 processException(e);
                 Thread.currentThread().interrupt();
@@ -701,7 +705,7 @@ void handleCommunicationIssue(Exception io) {
 
     // Close socket is called when another connect attempt is possible
     // Close is called when the connection should shut down, period
-    void closeSocket(boolean tryReconnectIfConnected) throws InterruptedException {
+    void closeSocket(boolean tryReconnectIfConnected, boolean forceClose) throws InterruptedException {
         // Ensure we close the socket exclusively within one thread.
         closeSocketLock.lock();
         try {
@@ -720,7 +724,7 @@ void closeSocket(boolean tryReconnectIfConnected) throws InterruptedException {
                 statusLock.unlock();
             }
 
-            closeSocketImpl();
+            closeSocketImpl(forceClose);
 
             statusLock.lock();
             try {
@@ -749,10 +753,10 @@ void closeSocket(boolean tryReconnectIfConnected) throws InterruptedException {
      */
     @Override
     public void close() throws InterruptedException {
-        this.close(true);
+        this.close(true, false);
     }
 
-    void close(boolean checkDrainStatus) throws InterruptedException {
+    void close(boolean checkDrainStatus, boolean forceClose) throws InterruptedException {
         statusLock.lock();
         try {
             if (checkDrainStatus && this.isDraining()) {
@@ -779,7 +783,7 @@ void close(boolean checkDrainStatus) throws InterruptedException {
             this.reconnectWaiter.cancel(true);
         }
 
-        closeSocketImpl();
+        closeSocketImpl(forceClose);
 
         this.dispatchers.forEach((nuid, d) -> d.stop(false));
 
@@ -831,7 +835,7 @@ void close(boolean checkDrainStatus) throws InterruptedException {
     }
 
     // Should only be called from closeSocket or close
-    void closeSocketImpl() {
+    void closeSocketImpl(boolean forceClose) {
         this.currentServer = null;
 
         // Signal both to stop.
@@ -854,8 +858,13 @@ void closeSocketImpl() {
 
         // Close the current socket and cancel anyone waiting for it
         try {
-            if (this.dataPort != null) {
-                this.dataPort.close();
+            if (dataPort != null) {
+                if (forceClose) {
+                    dataPort.forceClose();
+                }
+                else {
+                    dataPort.close();
+                }
             }
 
         } catch (IOException ex) {
@@ -2121,7 +2130,7 @@ public CompletableFuture<Boolean> drain(Duration timeout) throws TimeoutExceptio
         try {
             this.flush(timeout); // Flush and wait up to the timeout, if this fails, let the caller know
         } catch (Exception e) {
-            this.close(false);
+            this.close(false, false);
             throw e;
         }
 
@@ -2163,13 +2172,13 @@ public CompletableFuture<Boolean> drain(Duration timeout) throws TimeoutExceptio
                     }
                 }
 
-                this.close(false); // close the connection after the last flush
+                this.close(false, false); // close the connection after the last flush
                 tracker.complete(consumers.isEmpty());
             } catch (TimeoutException | InterruptedException e) {
                 this.processException(e);
             } finally {
                 try {
-                    this.close(false);// close the connection after the last flush
+                    this.close(false, false);// close the connection after the last flush
                 } catch (InterruptedException e) {
                     processException(e);
                     Thread.currentThread().interrupt();

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

@@ -25,6 +25,7 @@
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.net.SocketException;
 import java.net.URISyntaxException;
 import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
@@ -166,15 +167,10 @@ public void close() throws IOException {
     @Override
     public void forceClose() throws IOException {
         try {
-            // If we are here, and are being asked to force close,
-            // there is no need to linger. The dev might have set
-            // their own linger, in which case use theirs,
-            // otherwise set it to 0 for the quickest close
-            if (soLinger < 0) {
-                socket.setSoLinger(true, 0);
-            }
+            // If we are being asked to force close, there is no need to linger.
+            socket.setSoLinger(true, 0);
         }
-        catch (IOException e) {
+        catch (SocketException e) {
             // don't want to fail if I couldn't set linger
         }
         close();