allthingssecurity commented on code in PR #28320:
URL: https://github.com/apache/camel/pull/28320#discussion_r4358193811
##########
components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java:
##########
@@ -54,6 +54,11 @@ public class NatsConsumer extends DefaultConsumer {
private ExecutorService executor;
private Connection connection;
private Dispatcher dispatcher;
+ // the messages that the handler is processing, and the close of the
dispatcher of a stopped consumer 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 closeDispatcherWhenIdle;
Review Comment:
Done in 15aa72e0c6a5: `inFlightLock`, the in-flight count and the pending
close are now fields of `CamelNatsMessageHandler` (one per dispatcher). The
consumer keeps the handler next to `this.dispatcher` (`dispatcherHandler`,
volatile), and `doStop` calls `handler.closeWhenIdle(close)`, which closes now
or once that handler's message is done.
While writing the test I found that a route stop/start does not reuse the
consumer: Camel creates a new `NatsConsumer` on the route's next start, so
per-consumer state was already per run for that path. The same instance is
restarted in place when a route policy stops and starts a consumer that is not
`Suspendable` (for example `ThrottlingInflightRoutePolicy`, which does this
while exchanges are in flight). The new test
`eachRunOfTheConsumerClosesItsOwnDispatcher` does that with
`ServiceHelper.stopService/startService` on the route's consumer: stop and
start while a message of the first run is in flight, then stop again. The
second run's dispatcher must be closed at once, and the first one when its
message is done. Without the change it fails with `the dispatcher of the second
run is closed when it stops ==> expected: <1> but was: <0>`.
_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]
