Merge branch 'main' into put-locks-back

nats-io · Aug 20, 2024 · 5201214 · 5201214
2 parents 7fb3345 + 8806fbf
commit 5201214
Show file tree

Hide file tree

Showing 8 changed files with 24 additions and 4 deletions.
diff --git a/src/main/java/io/nats/client/impl/MessageQueue.java b/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();

diff --git a/src/main/java/io/nats/client/impl/NatsConnection.java b/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

diff --git a/src/main/java/io/nats/client/impl/NatsConnectionReader.java b/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

diff --git a/src/main/java/io/nats/client/impl/NatsConnectionWriter.java b/src/main/java/io/nats/client/impl/NatsConnectionWriter.java
@@ -211,8 +211,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);
         }

diff --git a/src/main/java/io/nats/client/impl/NatsConsumer.java b/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());
             }

diff --git a/src/main/java/io/nats/client/impl/NatsDispatcher.java b/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);

diff --git a/src/main/java/io/nats/client/impl/NatsDispatcherWithExecutor.java b/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);

diff --git a/src/main/java/io/nats/client/impl/NatsJetStreamImpl.java b/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);
         }
     }