Interrupt fine tuning (#1214)

Author: scottf

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

@@ -165,13 +165,7 @@ boolean push(NatsMessage msg, boolean internal) {
             if (!editLock.tryLock(offerLockMillis, TimeUnit.MILLISECONDS)) {
                 throw new IllegalStateException(OUTPUT_QUEUE_IS_FULL + queue.size());
             }
-        }
-        catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            throw new RuntimeException(e);
-        }
 
-        try {
             if (!internal && this.discardWhenFull) {
                 return this.queue.offer(msg);
             }
@@ -184,10 +178,12 @@ boolean push(NatsMessage msg, boolean internal) {
             this.sizeInBytes.getAndAdd(msg.getSizeInBytes());
             this.length.incrementAndGet();
             return true;
-        } catch (InterruptedException ie) {
+        }
+        catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             return false;
-        } finally {
+        }
+        finally {
             editLock.unlock();
         }
     }

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

@@ -320,19 +320,22 @@ void forceReconnectImpl(ForceReconnectOptions options) throws InterruptedExcepti
                             closeMe.close();
                         }
                     }
-                    catch (IOException ignore) {}
+                    catch (IOException ignore) {
+                    }
                 });
             }
 
             // stop i/o
             try {
                 this.reader.stop(false).get(100, TimeUnit.MILLISECONDS);
-            } catch (Exception ex) {
+            }
+            catch (Exception ex) {
                 processException(ex);
             }
             try {
                 this.writer.stop().get(100, TimeUnit.MILLISECONDS);
-            } catch (Exception ex) {
+            }
+            catch (Exception ex) {
                 processException(ex);
             }
 
@@ -344,19 +347,13 @@ void forceReconnectImpl(ForceReconnectOptions options) throws InterruptedExcepti
             closeSocketLock.unlock();
         }
 
-        try {
-            // calling connect just starts like a new connection versus reconnect
-            // but we have to manually resubscribe like reconnect once it is connected
-            reconnectImpl();
-            writer.setReconnectMode(false);
-        }
-        catch (InterruptedException e) {
-            // if there is an exception close() will have been called already
-            Thread.currentThread().interrupt();
-        }
+        // calling connect just starts like a new connection versus reconnect
+        // but we have to manually resubscribe like reconnect once it is connected
+        reconnectImpl();
+        writer.setReconnectMode(false);
     }
 
-   void reconnect() throws InterruptedException {
+    void reconnect() throws InterruptedException {
         if (!tryingToConnect.get()) {
             try {
                 tryingToConnect.set(true);

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

@@ -176,6 +176,8 @@ private List<Message> _fetch(int batchSize, long maxWaitMillis) {
             }
         }
         catch (InterruptedException e) {
+            // nextMessageInternal failed. By not throwing
+            // this gives them the messages already added to the list
             Thread.currentThread().interrupt();
         }
         return messages;
@@ -199,6 +201,8 @@ private List<Message> drainAlreadyBuffered(int batchSize) {
             }
         }
         catch (InterruptedException ignore) {
+            // nextMessageInternal failed. By not throwing
+            // this gives them the messages already added to the list
             Thread.currentThread().interrupt();
         }
         return messages;
@@ -275,7 +279,7 @@ public boolean hasNext() {
                         return false;
                     }
 
-                    if (buffered.size() == 0) {
+                    if (buffered.isEmpty()) {
                         msg = _nextUnmanaged(timeout, pullSubject);
                         if (msg == null) {
                             done = true;
@@ -292,6 +296,9 @@ public boolean hasNext() {
                 catch (InterruptedException e) {
                     msg = null;
                     done = true;
+                    // _nextUnmanaged failed
+                    // there still could be messages in the buffer
+                    // no good choice here
                     Thread.currentThread().interrupt();
                     return false;
                 }

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

@@ -47,6 +47,7 @@ public void run() {
                 }
                 catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
+                    // This task is going to re-run anyway, so no point in throwing
                 }
             }
         }

src/main/java/io/nats/service/Discovery.java

@@ -186,7 +186,9 @@ private byte[] discoverOne(String action, String serviceName, String serviceId)
             }
         }
         catch (InterruptedException e) {
+            // conn.request interrupted means data is not retrieved
             Thread.currentThread().interrupt();
+            throw new RuntimeException(e);
         }
         return null;
     }
@@ -215,7 +217,11 @@ private void discoverMany(String action, String serviceName, Consumer<byte[]> da
             }
         }
         catch (InterruptedException e) {
+            // sub.nextMessage was fetching one message
+            // and data is not completely read
+            // so it seems like this is an error condition
             Thread.currentThread().interrupt();
+            throw new RuntimeException(e);
         }
         finally {
             try {