Self review

scottf · scottf · commit 8dead5328bb9 · 2025-07-03T19:34:36.000-04:00
diff --git a/src/main/java/io/nats/client/impl/NatsFetchConsumer.java b/src/main/java/io/nats/client/impl/NatsFetchConsumer.java
@@ -67,7 +67,7 @@ public void pendingUpdated() {}
 
     @Override
     public void heartbeatError() {
-        finishAndShutdownSub();
+        fullClose();
     }
 
     @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
-                    finishAndShutdownSub();
+                    fullClose();
                 }
                 return m;
             }
@@ -103,15 +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
-                    finishAndShutdownSub();
+                    fullClose();
                 }
                 return m;
             }
 
             Message m = sub._nextUnmanaged(timeLeftNanos, pullSubject);
             if (m == null && isNoWaitNoExpires) {
                 // no message and no wait, go ahead and finish
-                finishAndShutdownSub();
+                fullClose();
             }
             return m;
         }
diff --git a/src/main/java/io/nats/client/impl/NatsMessageConsumer.java b/src/main/java/io/nats/client/impl/NatsMessageConsumer.java
@@ -53,7 +53,7 @@ class NatsMessageConsumer extends NatsMessageConsumerBase implements PullManager
     public void heartbeatError() {
         try {
             if (stopped.get()) {
-                finishAndShutdownSub();
+                fullClose();
             }
             else {
                 shutdownSub();
@@ -92,10 +92,10 @@ private void setupHbAlarmToTrigger() {
     public void pendingUpdated() {
         if (stopped.get()) {
             if (pmm.noMorePending()) {
-                finishAndShutdownSub();
+                fullClose();
             }
         }
-        else if ((pmm.pendingMessages <= thresholdMessages || (pmm.trackingBytes && pmm.pendingBytes <= thresholdBytes)))
+        else if (pmm.pendingMessages <= thresholdMessages || (pmm.trackingBytes && pmm.pendingBytes <= thresholdBytes))
         {
             repull();
         }
diff --git a/src/main/java/io/nats/client/impl/NatsMessageConsumerBase.java b/src/main/java/io/nats/client/impl/NatsMessageConsumerBase.java
@@ -105,7 +105,7 @@ public void close() throws Exception {
         shutdownSub();
     }
 
-    protected void finishAndShutdownSub() {
+    protected void fullClose() {
         stopped.set(true);
         finished.set(true);
         shutdownSub();

Original file line number	Diff line number	Diff line change
`@@ -67,7 +67,7 @@ public void pendingUpdated() {}`
`67`	`67`
`68`	`68`	`@Override`
`69`	`69`	`public void heartbeatError() {`
`70`		`- finishAndShutdownSub();`
	`70`	`+ fullClose();`
`71`	`71`	`}`
`72`	`72`
`73`	`73`	`@Override`
`@@ -85,7 +85,7 @@ public Message nextMessage() throws InterruptedException, JetStreamStatusChecked`
`85`	`85`	`if (m == null) {`
`86`	`86`	`// if there are no messages in the internal cache AND there are no more pending,`
`87`	`87`	`// they all have been read and we can go ahead and finish`
`88`		`- finishAndShutdownSub();`
	`88`	`+ fullClose();`
`89`	`89`	`}`
`90`	`90`	`return m;`
`91`	`91`	`}`
`@@ -103,15 +103,15 @@ public Message nextMessage() throws InterruptedException, JetStreamStatusChecked`
`103`	`103`	`Message m = sub._nextUnmanagedNoWait(pullSubject);`
`104`	`104`	`if (m == null) {`
`105`	`105`	`// no message and no time left, go ahead and finish`
`106`		`- finishAndShutdownSub();`
	`106`	`+ fullClose();`
`107`	`107`	`}`
`108`	`108`	`return m;`
`109`	`109`	`}`
`110`	`110`
`111`	`111`	`Message m = sub._nextUnmanaged(timeLeftNanos, pullSubject);`
`112`	`112`	`if (m == null && isNoWaitNoExpires) {`
`113`	`113`	`// no message and no wait, go ahead and finish`
`114`		`- finishAndShutdownSub();`
	`114`	`+ fullClose();`
`115`	`115`	`}`
`116`	`116`	`return m;`
`117`	`117`	`}`
Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,7 @@ class NatsMessageConsumer extends NatsMessageConsumerBase implements PullManager`
`53`	`53`	`public void heartbeatError() {`
`54`	`54`	`try {`
`55`	`55`	`if (stopped.get()) {`
`56`		`- finishAndShutdownSub();`
	`56`	`+ fullClose();`
`57`	`57`	`}`
`58`	`58`	`else {`
`59`	`59`	`shutdownSub();`
`@@ -92,10 +92,10 @@ private void setupHbAlarmToTrigger() {`
`92`	`92`	`public void pendingUpdated() {`
`93`	`93`	`if (stopped.get()) {`
`94`	`94`	`if (pmm.noMorePending()) {`
`95`		`- finishAndShutdownSub();`
	`95`	`+ fullClose();`
`96`	`96`	`}`
`97`	`97`	`}`
`98`		`- else if ((pmm.pendingMessages <= thresholdMessages \|\| (pmm.trackingBytes && pmm.pendingBytes <= thresholdBytes)))`
	`98`	`+ else if (pmm.pendingMessages <= thresholdMessages \|\| (pmm.trackingBytes && pmm.pendingBytes <= thresholdBytes))`
`99`	`99`	`{`
`100`	`100`	`repull();`
`101`	`101`	`}`
Original file line number	Diff line number	Diff line change
`@@ -105,7 +105,7 @@ public void close() throws Exception {`
`105`	`105`	`shutdownSub();`
`106`	`106`	`}`
`107`	`107`
`108`		`- protected void finishAndShutdownSub() {`
	`108`	`+ protected void fullClose() {`
`109`	`109`	`stopped.set(true);`
`110`	`110`	`finished.set(true);`
`111`	`111`	`shutdownSub();`