davsclaus commented on code in PR #28320:
URL: https://github.com/apache/camel/pull/28320#discussion_r4367325393
##########
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:
Added the comment line in e1cde596ac8d.
_Claude Code on behalf of davsclaus_
--
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]
