Merge pull request #1412 from nats-io/write-queue-info

Expose connection outgoingPendingMessageCount and outgoingPendingBytes

Author: scottf

src/main/java/io/nats/client/Connection.java

@@ -785,4 +785,22 @@ enum Status {
      */
     @NonNull
     ObjectStoreManagement objectStoreManagement(ObjectStoreOptions options) throws IOException;
+
+    /**
+     * Get the number of messages in the outgoing queue for this connection.
+     * This value is volatile in the sense that it changes often and may be adjusted by more than one message.
+     * It changes every time a message is published (put in the outgoing queue)
+     * and every time a message is removed from the queue to be written over the socket
+     * @return the number of messages in the outgoing queue
+     */
+    long outgoingPendingMessageCount();
+
+    /**
+     * Get the number of bytes based to be written calculated from the messages in the outgoing queue for this connection.
+     * This value is volatile in the sense that it changes often and may be adjusted by more than one message's bytes.
+     * It changes every time a message is published (put in the outgoing queue)
+     * and every time a message is removed from the queue to be written over the socket
+     * @return the number of messages in the outgoing queue
+     */
+    long outgoingPendingBytes();
 }

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

@@ -133,7 +133,7 @@ void drain() {
 
     boolean isDrained() {
         // poison pill is not included in the length count, or the size
-        return this.running.get() == DRAINING && this.length() == 0;
+        return this.running.get() == DRAINING && this.length.get() == 0;
     }
 
     boolean push(NatsMessage msg) {
@@ -326,7 +326,6 @@ NatsMessage popNow() throws InterruptedException {
         return pop(null);
     }
 
-    // Just for testing
     long length() {
         return this.length.get();
     }

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

@@ -2544,4 +2544,20 @@ private void ensureNotClosing() throws IOException {
             throw new IOException("A JetStream context can't be established during close.");
         }
     }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public long outgoingPendingMessageCount() {
+        return writer.outgoingPendingMessageCount();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public long outgoingPendingBytes() {
+        return writer.outgoingPendingBytes();
+    }
 }

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

@@ -254,4 +254,12 @@ void flushBuffer() {
             writerLock.unlock();
         }
     }
+
+    long outgoingPendingMessageCount() {
+        return outgoing.length();
+    }
+
+    long outgoingPendingBytes() {
+        return outgoing.sizeInBytes();
+    }
 }