gnodet-bot commented on code in PR #28320:
URL: https://github.com/apache/camel/pull/28320#discussion_r4365084270
##########
components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java:
##########
@@ -551,9 +603,50 @@ class CamelNatsMessageHandler implements MessageHandler {
final boolean ackPolicyNone = configuration.getAckPolicy() ==
AckPolicy.None;
final boolean manualAckEnabled = configuration.isManualAck() &&
configuration.isJetstreamEnabled();
+ // the messages that this handler is processing, and the close of
its dispatcher (the consumer stopped)
+ // that waits until none is: closing a dispatcher interrupts its
thread, which runs the route
+ private final Object inFlightLock = new Object();
+ private int messagesInFlight;
+ private Runnable pendingClose;
+
+ /**
+ * Closes the dispatcher of this handler now, or once the message
in flight is done.
+ */
+ void closeWhenIdle(Runnable close) {
+ synchronized (inFlightLock) {
+ if (messagesInFlight > 0) {
+ // closed by onMessage once the message is done
+ pendingClose = close;
+ return;
+ }
+ }
+ close.run();
Review Comment:
💡 **Nit (from earlier review, still open):** `close.run()` is called outside
the `synchronized` block intentionally — the close may block (it closes the
dispatcher thread). A one-liner would help future readers:
```suggestion
}
// no message in flight: close now (outside the lock, as the
close may block)
close.run();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
