Merge pull request #1416 from nats-io/lock-around-outgoing

Lock around access to pending message and byte counts

Author: scottf

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

@@ -54,7 +54,7 @@ class NatsConnection implements Connection {
     private boolean disconnecting; // you can only disconnect in one thread
     private boolean closing; // respect a close call regardless
     private Exception exceptionDuringConnectChange; // exception occurred in another thread while dis/connecting
-    final ReentrantLock closeSocketLock;
+    final ReentrantLock closeSocketLock; // this is not private so it can be tested
 
     private Status status;
     private final ReentrantLock statusLock;
@@ -2550,14 +2550,26 @@ private void ensureNotClosing() throws IOException {
      */
     @Override
     public long outgoingPendingMessageCount() {
-        return writer.outgoingPendingMessageCount();
+        closeSocketLock.lock();
+        try {
+            return writer == null ? -1 : writer.outgoingPendingMessageCount();
+        }
+        finally {
+            closeSocketLock.unlock();
+        }
     }
 
     /**
      * {@inheritDoc}
      */
     @Override
     public long outgoingPendingBytes() {
-        return writer.outgoingPendingBytes();
+        closeSocketLock.lock();
+        try {
+            return writer == null ? -1 : writer.outgoingPendingBytes();
+        }
+        finally {
+            closeSocketLock.unlock();
+        }
     }
 }

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

@@ -256,10 +256,22 @@ void flushBuffer() {
     }
 
     long outgoingPendingMessageCount() {
-        return outgoing.length();
+        writerLock.lock();
+        try {
+            return outgoing == null ? -1 : outgoing.length();
+        }
+        finally {
+            writerLock.unlock();
+        }
     }
 
     long outgoingPendingBytes() {
-        return outgoing.sizeInBytes();
+        writerLock.lock();
+        try {
+            return outgoing == null ? -1 : outgoing.length();
+        }
+        finally {
+            writerLock.unlock();
+        }
     }
 }