allthingssecurity commented on code in PR #27919:
URL: https://github.com/apache/camel/pull/27919#discussion_r4291295742
##########
components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java:
##########
@@ -68,24 +91,125 @@ private long processInitialEvent() {
@Override
protected void doStart() throws Exception {
- this.eventsCmd = DockerClientFactory.getDockerClient(component,
endpoint.getConfiguration(), null).eventsCmd()
- .withSince(String.valueOf(processInitialEvent()));
- this.eventsCmd.exec(new EventsCallback());
+ executor =
getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
+ "DockerEventsConsumer");
+ try {
+ subscribe(String.valueOf(processInitialEvent()),
SUBSCRIBE_AGAIN_DELAY_MILLIS);
+ } catch (Exception e) {
+ // a failed start does not call doStop()
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
+ executor = null;
+ throw e;
+ }
super.doStart();
}
@Override
protected void doStop() throws Exception {
- this.eventsCmd.close();
+ if (executor != null) {
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
+ executor = null;
+ }
+ EventsCallback current = callback;
+ callback = null;
+ if (current != null) {
+ current.close();
+ }
+ if (eventsCmd != null) {
+ eventsCmd.close();
+ }
super.doStop();
}
+ /**
+ * Opens a new event stream.
+ *
+ * @param since the value of the since parameter, or <tt>null</tt> to
receive the events from now on
+ * @param retryDelay the delay before subscribing again if this stream
fails before it is open
+ */
+ private void subscribe(String since, long retryDelay) {
+ EventsCmd cmd = DockerClientFactory.getDockerClient(component,
endpoint.getConfiguration(), null).eventsCmd();
+ if (since != null) {
+ cmd = cmd.withSince(since);
+ }
+ EventsCallback newCallback = new EventsCallback(retryDelay);
+ this.eventsCmd = cmd;
+ this.callback = newCallback;
+ cmd.exec(newCallback);
+ if (!isRunAllowed()) {
+ // the consumer was stopped while it subscribed again, so stopping
could not close this stream
+ callback = null;
+ newCallback.closeQuietly();
+ }
+ }
+
+ /**
+ * Subscribes again after the delay, as long as the consumer runs. The
events of the time without a stream are not
+ * received: the new stream starts with the events from then on.
+ */
+ private void subscribeAgain(long delayMillis) {
+ ScheduledExecutorService current = executor;
+ if (!isRunAllowed() || current == null) {
+ return;
+ }
+ try {
+ current.schedule(() -> {
+ if (!isRunAllowed()) {
+ return;
+ }
+ long nextDelay = Math.min(delayMillis * 2,
SUBSCRIBE_AGAIN_MAX_DELAY_MILLIS);
+ try {
+ subscribe(null, nextDelay);
+ } catch (Exception e) {
+ LOG.warn("Failed to subscribe again to the Docker events
of {}, trying again in {} ms: {}",
+ endpoint, nextDelay, e.getMessage(), e);
Review Comment:
Thanks, done in ebd724e5a110: a failed attempt to subscribe again now logs
one WARN line with the message, and the stack trace at DEBUG, as you suggested.
camel-docker tests: 54 run, 0 failures, 4 skipped.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java:
##########
@@ -36,9 +42,26 @@ public class DockerEventsConsumer extends DefaultConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(DockerEventsConsumer.class);
+ /**
+ * The delay before the consumer subscribes again after its event stream
failed or ended.
+ */
+ static final long SUBSCRIBE_AGAIN_DELAY_MILLIS = 1000;
+
+ /**
+ * The maximum delay between two attempts to subscribe again, when
subscribing keeps failing (for example while the
+ * Docker daemon cannot be reached).
+ */
+ static final long SUBSCRIBE_AGAIN_MAX_DELAY_MILLIS = 30000;
+
private DockerEndpoint endpoint;
private DockerComponent component;
- private EventsCmd eventsCmd;
+ private volatile EventsCmd eventsCmd;
+ /**
+ * The callback of the current event stream. docker-java opens the stream
on its own thread and the stream belongs
+ * to this callback: closing the command does not close it.
+ */
+ private volatile EventsCallback callback;
+ private ScheduledExecutorService executor;
Review Comment:
Done in ebd724e5a110: `executor` is now `volatile`, like `eventsCmd` and
`callback`.
_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]
