finishAndClose();

Author: scottf

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

@@ -67,7 +67,7 @@ public void pendingUpdated() {}
 
     @Override
     public void heartbeatError() {
-        finish(true);
+        finishAndClose();
     }
 
     @Override
@@ -85,7 +85,7 @@ public Message nextMessage() throws InterruptedException, JetStreamStatusChecked
                 if (m == null) {
                     // if there are no messages in the internal cache AND there are no more pending,
                     // they all have been read and we can go ahead and finish
-                    finish(true);
+                    finishAndClose();
                 }
                 return m;
             }
@@ -103,16 +103,15 @@ public Message nextMessage() throws InterruptedException, JetStreamStatusChecked
                 Message m = sub._nextUnmanagedNoWait(pullSubject);
                 if (m == null) {
                     // no message and no time left, go ahead and finish
-                    finish(true);
+                    finishAndClose();
                 }
                 return m;
             }
 
             Message m = sub._nextUnmanaged(timeLeftNanos, pullSubject);
             if (m == null && isNoWaitNoExpires) {
                 // no message and no wait, go ahead and finish
-                finished.set(true);
-                lenientClose();
+                finishAndClose();
             }
             return m;
         }

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

@@ -55,7 +55,7 @@ public void heartbeatError() {
             // just close the current sub and make another one.
             // this could go on endlessly - unless the user had called stop
             if (stopped.get()) {
-                finish(true);
+                finishAndClose();
             }
             else {
                 lenientClose();
@@ -72,7 +72,7 @@ void doSub() throws JetStreamApiException, IOException {
         MessageHandler mh = userMessageHandler == null ? null : msg -> {
             userMessageHandler.onMessage(msg);
             if (stopped.get() && pmm.noMorePending()) {
-                finish(false);
+                finishAndClose();
             }
         };
         super.initSub(subscriptionMaker.subscribe(mh, userDispatcher, pmm, null));
@@ -85,7 +85,7 @@ void doSub() throws JetStreamApiException, IOException {
     public void pendingUpdated() {
         if (stopped.get()) {
             if (pmm.noMorePending()) {
-                finish(false);
+                finishAndClose();
             }
         }
         else if (pmm.pendingMessages <= thresholdMessages || (pmm.trackingBytes && pmm.pendingBytes <= thresholdBytes))

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

@@ -104,14 +104,12 @@ public void close() throws Exception {
         lenientClose();
     }
 
-    protected void finish(boolean close) {
-        finished.set(true);
+    protected void finishAndClose() {
         if (pmm != null) {
             pmm.shutdownHeartbeatTimer();
         }
-        if (close) {
-            lenientClose();
-        }
+        finished.set(true);
+        lenientClose();
     }
 
     protected void lenientClose() {