Merge branch 'main' into wbl3

scottf · scottf · commit 345825aa419e · 2025-10-24T12:45:55.000-04:00
# Conflicts:
#	src/main/java/io/nats/client/impl/NatsConnectionWriter.java
diff --git a/src/main/java/io/nats/client/StatisticsCollector.java b/src/main/java/io/nats/client/StatisticsCollector.java
@@ -84,6 +84,7 @@ public interface StatisticsCollector extends Statistics {
     /**
      * Increments the total number of messages that have come in to this connection
      * by 1 AND the number of bytes in the same call.
+     * @param bytes the number of bytes
      */
     default void incrementIn(long bytes) {
         incrementInMsgs();
@@ -93,6 +94,7 @@ default void incrementIn(long bytes) {
     /**
      * Increments the total number of messages that have gone out of this connection.
      * by 1 AND the number of bytes in the same call.
+     * @param bytes the number of bytes
      */
     default void incrementOut(long bytes) {
         incrementOutMsgs();
diff --git a/src/main/java/io/nats/client/impl/MessageQueue.java b/src/main/java/io/nats/client/impl/MessageQueue.java
@@ -79,8 +79,8 @@ class MessageQueue {
         this.queue = maxMessagesInOutgoingQueue > 0 ? new LinkedBlockingQueue<>(maxMessagesInOutgoingQueue) : new LinkedBlockingQueue<>();
         this.discardWhenFull = discardWhenFull;
         this.running = new AtomicInteger(RUNNING);
-        this.sizeInBytes = new AtomicLong(0);
-        this.length = new AtomicLong(0);
+        sizeInBytes = new AtomicLong(0);
+        length = new AtomicLong(0);
         this.offerLockNanos = requestCleanupInterval.toNanos();
         this.offerTimeoutNanos = Math.max(MIN_OFFER_TIMEOUT_NANOS, requestCleanupInterval.toMillis() * NANOS_PER_MILLI * 95 / 100) ;
 
@@ -98,8 +98,8 @@ void drainTo(MessageQueue target) {
         editLock.lock();
         try {
             queue.drainTo(target.queue);
-            target.length.set(length.get());
-            target.sizeInBytes.set(sizeInBytes.get());
+            target.length.set(length.getAndSet(0));
+            target.sizeInBytes.set(sizeInBytes.getAndSet(0));
         } finally {
             editLock.unlock();
         }
@@ -132,8 +132,8 @@ void drain() {
     }
 
     boolean isDrained() {
-        // poison pill is not included in the length count, or the size
-        return this.running.get() == DRAINING && this.length.get() == 0;
+        // poison pill  and any other "mark" messages are not included in the length count, or the size
+        return this.running.get() == DRAINING && length.get() == 0;
     }
 
     boolean push(NatsMessage msg) {
@@ -165,17 +165,19 @@ boolean push(NatsMessage msg, boolean internal) {
              */
             if (editLock.tryLock(offerLockNanos, TimeUnit.NANOSECONDS)) {
                 try {
-                    if (!internal && this.discardWhenFull) {
-                        return this.queue.offer(msg);
+                    // offer with no timeout returns true if the queue was not full
+                    if (!internal && discardWhenFull) {
+                        return queue.offer(msg);
                     }
 
                     long timeoutNanosLeft = Math.max(MIN_OFFER_TIMEOUT_NANOS, offerTimeoutNanos - (NatsSystemClock.nanoTime() - startNanos));
 
-                    if (!this.queue.offer(msg, timeoutNanosLeft, TimeUnit.NANOSECONDS)) {
+                    // offer with timeout
+                    if (!queue.offer(msg, timeoutNanosLeft, TimeUnit.NANOSECONDS)) {
                         throw new IllegalStateException(OUTPUT_QUEUE_IS_FULL + queue.size());
                     }
-                    this.sizeInBytes.getAndAdd(msg.getSizeInBytes());
-                    this.length.incrementAndGet();
+                    sizeInBytes.getAndAdd(msg.getSizeInBytes());
+                    length.incrementAndGet();
                     return true;
 
                 }
@@ -196,13 +198,20 @@ boolean push(NatsMessage msg, boolean internal) {
     /**
      * poisoning the queue puts the known poison pill into the queue, forcing any waiting code to stop
      * waiting and return.
+     * This is done outside of push so it is not counted in length or size
+     * offer is used instead of add since we don't care if it fails because it was full
      */
     void poisonTheQueue() {
-        try {
-            this.queue.add(POISON_PILL);
-        } catch (IllegalStateException ie) { // queue was full, so we don't really need poison pill
-            // ok to ignore this
-        }
+        queue.offer(POISON_PILL);
+    }
+
+    /**
+     * Marking the queue, like POISON, is a message we don't want to count.
+     * Intended to only be used with an unbounded queue. Use at your own risk.
+     * @param msg the mark
+     */
+    void markTheQueue(NatsMessage msg) {
+        queue.offer(msg);
     }
 
     NatsMessage poll(Duration timeout) throws InterruptedException {
@@ -240,8 +249,8 @@ NatsMessage pop(Duration timeout) throws InterruptedException {
             return null;
         }
 
-        this.sizeInBytes.getAndAdd(-msg.getSizeInBytes());
-        this.length.decrementAndGet();
+        sizeInBytes.getAndAdd(-msg.getSizeInBytes());
+        length.decrementAndGet();
 
         return msg;
     }
@@ -276,12 +285,12 @@ NatsMessage accumulate(long maxBytesToAccumulate, long maxMessagesToAccumulate,
         long size = msg.getSizeInBytes();
 
         if (maxMessagesToAccumulate <= 1 || size >= maxBytesToAccumulate) {
-            this.sizeInBytes.addAndGet(-size);
-            this.length.decrementAndGet();
+            sizeInBytes.addAndGet(-size);
+            length.decrementAndGet();
             return msg;
         }
 
-        long count = 1;
+        long accumulated = 1;
         NatsMessage cursor = msg;
 
         while (true) {
@@ -290,15 +299,15 @@ NatsMessage accumulate(long maxBytesToAccumulate, long maxMessagesToAccumulate,
                 long s = next.getSizeInBytes();
                 if (maxBytesToAccumulate < 0 || (size + s) < maxBytesToAccumulate) { // keep going
                     size += s;
-                    count++;
+                    accumulated++;
 
                     this.queue.poll(); // we need to get the message out of the queue b/c we only peeked
                     cursor.next = next;
                     if (next.flushImmediatelyAfterPublish) {
                         // if we are going to flush, then don't accumulate more
                         break;
                     }
-                    if (count == maxMessagesToAccumulate) {
+                    if (accumulated == maxMessagesToAccumulate) {
                         break;
                     }
                     cursor = cursor.next;
@@ -310,8 +319,8 @@ NatsMessage accumulate(long maxBytesToAccumulate, long maxMessagesToAccumulate,
             }
         }
 
-        this.sizeInBytes.addAndGet(-size);
-        this.length.addAndGet(-count);
+        sizeInBytes.addAndGet(-size);
+        length.addAndGet(-accumulated);
 
         return msg;
     }
@@ -322,11 +331,11 @@ NatsMessage popNow() throws InterruptedException {
     }
 
     long length() {
-        return this.length.get();
+        return length.get();
     }
 
     long sizeInBytes() {
-        return this.sizeInBytes.get();
+        return sizeInBytes.get();
     }
 
     void filter(Predicate<NatsMessage> p) {
@@ -341,8 +350,8 @@ void filter(Predicate<NatsMessage> p) {
                 if (!p.test(cursor)) {
                     newQueue.add(cursor);
                 } else {
-                    this.sizeInBytes.addAndGet(-cursor.getSizeInBytes());
-                    this.length.decrementAndGet();
+                    sizeInBytes.addAndGet(-cursor.getSizeInBytes());
+                    length.decrementAndGet();
                 }
                 cursor = this.queue.poll();
             }
@@ -356,8 +365,8 @@ void clear() {
         editLock.lock();
         try {
             this.queue.clear();
-            this.length.set(0);
-            this.sizeInBytes.set(0);
+            length.set(0);
+            sizeInBytes.set(0);
         } finally {
             editLock.unlock();
         }
diff --git a/src/main/java/io/nats/client/impl/NatsConnectionWriter.java b/src/main/java/io/nats/client/impl/NatsConnectionWriter.java
@@ -1,4 +1,5 @@
-// Copyright 2015-2025 The NATS Authors
+
+// Copyright 2015-2018 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at:
@@ -242,7 +243,7 @@ void enterReconnectMode() {
     }
 
     void enterWaitingForEndReconnectMode() {
-        reconnectOutgoing.push(END_RECONNECT);
+        reconnectOutgoing.markTheQueue(END_RECONNECT);
         mode.set(Mode.WaitingForEndReconnect);
     }
 
diff --git a/src/test/java/io/nats/client/impl/MessageQueueTests.java b/src/test/java/io/nats/client/impl/MessageQueueTests.java

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,5 @@`
`1`		`-// Copyright 2015-2025 The NATS Authors`
	`1`	`+`
	`2`	`+// Copyright 2015-2018 The NATS Authors`
`2`	`3`	`// Licensed under the Apache License, Version 2.0 (the "License");`
`3`	`4`	`// you may not use this file except in compliance with the License.`
`4`	`5`	`// You may obtain a copy of the License at:`
`@@ -242,7 +243,7 @@ void enterReconnectMode() {`
`242`	`243`	`}`
`243`	`244`
`244`	`245`	`void enterWaitingForEndReconnectMode() {`
`245`		`- reconnectOutgoing.push(END_RECONNECT);`
	`246`	`+ reconnectOutgoing.markTheQueue(END_RECONNECT);`
`246`	`247`	`mode.set(Mode.WaitingForEndReconnect);`
`247`	`248`	`}`
`248`	`249`