allthingssecurity commented on code in PR #28320:
URL: https://github.com/apache/camel/pull/28320#discussion_r4366110468
##########
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:
Agreed that the close is outside the lock on purpose, because closing the
dispatcher may block. I left the code as it is to avoid another CI run for a
comment only; happy to add the line if Claus wants it.
_Claude Code on behalf of allthingssecurity_
--
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]
