When interrupted call interrupt

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

Author: MauriceVanVeen

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

@@ -185,6 +185,7 @@ boolean push(NatsMessage msg, boolean internal) {
             this.length.incrementAndGet();
             return true;
         } catch (InterruptedException ie) {
+            Thread.currentThread().interrupt();
             return false;
         } finally {
             editLock.unlock();

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

@@ -653,6 +653,7 @@ public void run() {
                 this.closeSocket(false, true);
             } catch (InterruptedException e) {
                 processException(e);
+                Thread.currentThread().interrupt();
             }
         } finally {
             statusLock.lock();
@@ -1516,7 +1517,11 @@ public Duration RTT() throws IOException {
         catch (ExecutionException e) {
             throw new IOException(e.getCause());
         }
-        catch (InterruptedException | TimeoutException e) {
+        catch (TimeoutException e) {
+            throw new IOException(e);
+        }
+        catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             throw new IOException(e);
         }
     }
@@ -2211,8 +2216,11 @@ public CompletableFuture<Boolean> drain(Duration timeout) throws TimeoutExceptio
 
                 this.close(false, false); // close the connection after the last flush
                 tracker.complete(consumers.isEmpty());
-            } catch (TimeoutException | InterruptedException e) {
+            } catch (TimeoutException e) {
+                this.processException(e);
+            } catch (InterruptedException e) {
                 this.processException(e);
+                Thread.currentThread().interrupt();
             } finally {
                 try {
                     this.close(false, false);// close the connection after the last flush

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

@@ -172,8 +172,11 @@ else if (this.mode == Mode.GATHER_HEADERS) {
             if (running.get()) {
                 this.connection.handleCommunicationIssue(io);
             }
-        } catch (CancellationException | ExecutionException | InterruptedException ex) {
+        } catch (CancellationException | ExecutionException ex) {
             // Exit
+        } catch (InterruptedException ex) {
+            // Exit
+            Thread.currentThread().interrupt();
         } finally {
             this.running.set(false);
             // Clear the buffers, since they are only used inside this try/catch

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

@@ -209,8 +209,11 @@ public void run() {
             if (running.get()) {
                 this.connection.handleCommunicationIssue(io);
             }
-        } catch (CancellationException | ExecutionException | InterruptedException ex) {
+        } catch (CancellationException | ExecutionException ex) {
             // Exit
+        } catch (InterruptedException ex) {
+            // Exit
+            Thread.currentThread().interrupt();
         } finally {
             this.running.set(false);
         }

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

@@ -221,6 +221,7 @@ public CompletableFuture<Boolean> drain(Duration timeout) throws InterruptedExce
                 this.cleanUpAfterDrain();
             } catch (InterruptedException e) {
                 this.connection.processException(e);
+                Thread.currentThread().interrupt();
             } finally {
                 tracker.complete(this.isDrained());
             }

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

@@ -126,6 +126,7 @@ public void run() {
             if (this.running.get()){
                 this.connection.processException(exp);
             } //otherwise we did it
+            Thread.currentThread().interrupt();
         }
         finally {
             this.running.set(false);

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

@@ -69,6 +69,7 @@ public void run() {
             if (this.running.get()){
                 this.connection.processException(exp);
             } //otherwise we did it
+            Thread.currentThread().interrupt();
         }
         finally {
             this.running.set(false);

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

@@ -232,6 +232,7 @@ Message makeRequestResponseRequired(String subject, byte[] bytes, Duration timeo
         try {
             return responseRequired(conn.request(prependPrefix(subject), bytes, timeout));
         } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             throw new IOException(e);
         }
     }
@@ -240,6 +241,7 @@ Message makeInternalRequestResponseRequired(String subject, Headers headers, byt
         try {
             return responseRequired(conn.requestInternal(subject, headers, data, timeout, cancelAction, validateSubjectAndReplyTo));
         } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             throw new IOException(e);
         }
     }