This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 22e3ffa04972 CAMEL-26688: camel-core - step listener SPI for the
exchange sanitizer (#29374)
22e3ffa04972 is described below
commit 22e3ffa0497230da6c98f178a5b42dd8972ac675
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Nov 7 20:19:37 2026 +0530
CAMEL-26688: camel-core - step listener SPI for the exchange sanitizer
(#29374)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../main/java/org/apache/camel/AsyncCallback.java | 8 +-
.../main/java/org/apache/camel/AsyncProcessor.java | 30 +-
.../org/apache/camel/ExtendedCamelContext.java | 37 ++
.../java/org/apache/camel/spi/StepListener.java | 118 +++++
.../java/org/apache/camel/spi/Synchronization.java | 20 +
.../impl/engine/DefaultCamelContextExtension.java | 22 +
.../apache/camel/impl/engine/DefaultChannel.java | 9 +
.../camel/impl/engine/StepListenerProcessor.java | 198 ++++++++
.../StepListenerInterceptStrategyTest.java | 213 ++++++++
.../processor/StepListenerNotInstalledTest.java | 95 ++++
.../camel/processor/StepListenerRecorder.java | 112 +++++
.../apache/camel/processor/StepListenerTest.java | 539 +++++++++++++++++++++
.../camel/management/ManagedStepListenerTest.java | 171 +++++++
design/exchange-sanitizer.adoc | 2 +-
14 files changed, 1567 insertions(+), 7 deletions(-)
diff --git a/core/camel-api/src/main/java/org/apache/camel/AsyncCallback.java
b/core/camel-api/src/main/java/org/apache/camel/AsyncCallback.java
index 2692c50a26c6..5256468c267f 100644
--- a/core/camel-api/src/main/java/org/apache/camel/AsyncCallback.java
+++ b/core/camel-api/src/main/java/org/apache/camel/AsyncCallback.java
@@ -32,9 +32,15 @@ public interface AsyncCallback extends Runnable {
* <p/>
* If an exception occurred while processing the exchange, the exception
field of the {@link Exchange} being
* processed will hold the caused exception.
+ * <p/>
+ * It is called exactly once for each call of {@link
AsyncProcessor#process(Exchange, AsyncCallback)} it was given
+ * to, also when the processing failed, and never a second time: a second
call can continue an exchange that was
+ * already continued. See the contract in {@link AsyncProcessor}.
*
* @param doneSync is <tt>true</tt> if the processing of the {@link
Exchange} was completed by a synchronous thread.
- * Otherwise, its <tt>false</tt> to indicate it was
completed by an asynchronous thread.
+ * Otherwise, its <tt>false</tt> to indicate it was
completed by an asynchronous thread. It is the
+ * value that {@link AsyncProcessor#process(Exchange,
AsyncCallback)} returns, except for a
+ * processor that hands its callback to the reactive
executor (see {@link AsyncProcessor}).
*/
void done(boolean doneSync);
diff --git a/core/camel-api/src/main/java/org/apache/camel/AsyncProcessor.java
b/core/camel-api/src/main/java/org/apache/camel/AsyncProcessor.java
index e2db9d1f2634..a24c7a43bb5d 100644
--- a/core/camel-api/src/main/java/org/apache/camel/AsyncProcessor.java
+++ b/core/camel-api/src/main/java/org/apache/camel/AsyncProcessor.java
@@ -24,6 +24,24 @@ import java.util.concurrent.CompletableFuture;
* <p/>
* Any processor can be coerced to have an {@link AsyncProcessor} interface by
using the
* {@link org.apache.camel.support.AsyncProcessorConverterHelper#convert
AsyncProcessorConverterHelper.convert} method.
+ * <p/>
+ * <b>The contract of {@link #process(Exchange, AsyncCallback)}.</b> The
routing engine relies on it, and a processor
+ * that breaks the first two rules is a bug, whether it is Camel's or a user's:
+ * <ul>
+ * <li>The callback is called exactly once for each call of {@code process}:
also on a failure, and never a second time
+ * (for example once from a timeout and again from a reply that comes late). A
second call can continue an exchange that
+ * was already continued.</li>
+ * <li>{@code process} does not throw. On a failure it sets the exception on
the exchange
+ * ({@link Exchange#setException(Throwable)}) and calls the callback.</li>
+ * <li>The value the callback is called with ({@code doneSync}) is the value
{@code process} returns. When
+ * {@code process} returns {@code true}, the callback has been called already,
on the calling thread; when it returns
+ * {@code false}, the callback is called with {@code false}, possibly on
another thread and possibly before
+ * {@code process} returns. A processor that hands its callback to the {@link
org.apache.camel.spi.ReactiveExecutor}
+ * instead of calling it, as the channel of each step of a route does, is the
exception: it returns the value of the
+ * processor it calls, and the reactive executor runs its callback with {@code
false}, at once or after the work the
+ * thread is running. So a caller of such a processor must go on through the
callback, not through the return
+ * value.</li>
+ * </ul>
*
* @see Processor
* @see AsyncCallback
@@ -34,14 +52,16 @@ public interface AsyncProcessor extends Processor {
* Processes the message exchange. Similar to {@link Processor#process},
but the caller supports having the exchange
* asynchronously processed.
* <p/>
- * If there was a failure processing then the caused {@link Exception}
would be set on the {@link Exchange}.
+ * If there was a failure processing then the caused {@link Exception}
would be set on the {@link Exchange}, and the
+ * callback is called; this method does not throw. See the contract in the
class Javadoc.
*
* @param exchange the message exchange
- * @param callback the {@link AsyncCallback} will be invoked when the
processing of the exchange is completed. If
- * the exchange is completed synchronously, then the
callback is also invoked synchronously. The
- * callback should therefore be careful of starting
recursive loop.
+ * @param callback the {@link AsyncCallback} will be invoked exactly once
when the processing of the exchange is
+ * completed. If the exchange is completed synchronously,
then the callback is also invoked
+ * synchronously. The callback should therefore be
careful of starting recursive loop.
* @return (doneSync) <tt>true</tt> to continue execution
synchronously, <tt>false</tt> to continue being
- * executed asynchronously
+ * executed asynchronously; the same value the callback
is called with, except for a processor that
+ * hands its callback to the reactive executor (see above)
*/
boolean process(Exchange exchange, AsyncCallback callback);
diff --git
a/core/camel-api/src/main/java/org/apache/camel/ExtendedCamelContext.java
b/core/camel-api/src/main/java/org/apache/camel/ExtendedCamelContext.java
index dc72deee55a7..a703b5e0d01b 100644
--- a/core/camel-api/src/main/java/org/apache/camel/ExtendedCamelContext.java
+++ b/core/camel-api/src/main/java/org/apache/camel/ExtendedCamelContext.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -45,6 +46,7 @@ import org.apache.camel.spi.RouteStartupOrder;
import org.apache.camel.spi.SendHook;
import org.apache.camel.spi.StartupStepRecorder;
import org.apache.camel.spi.StepHook;
+import org.apache.camel.spi.StepListener;
import org.jspecify.annotations.Nullable;
/**
@@ -541,6 +543,41 @@ public interface ExtendedCamelContext {
throw new UnsupportedOperationException("This CamelContext does not
support a StepHook");
}
+ /**
+ * Adds a {@link StepListener} that observes each call of the processor of
a step of a route. Many listeners can be
+ * installed; adding a listener that is already installed (equal to one
installed) is ignored. Whether a route calls
+ * the listeners is decided when it is created, so add them before the
routes are created: a route created while no
+ * listener was installed is not seen. The CamelContext does not start or
stop the listeners, and they stay
+ * installed when the CamelContext is stopped and started again.
+ *
+ * @param listener the listener
+ * @since 4.24
+ */
+ default void addStepListener(StepListener listener) {
+ throw new UnsupportedOperationException("This CamelContext does not
support a StepListener");
+ }
+
+ /**
+ * Removes a {@link StepListener} (one equal to it). A call of a step that
started before keeps the listener for its
+ * callback.
+ *
+ * @param listener the listener
+ * @since 4.24
+ */
+ default void removeStepListener(StepListener listener) {
+ // no listener can be installed, so there is nothing to remove
+ }
+
+ /**
+ * Gets the {@link StepListener}s, in the order they were added.
+ *
+ * @return an unmodifiable view of the listeners (empty if none is
installed), which reflects later adds and removes
+ * @since 4.24
+ */
+ default List<StepListener> getStepListeners() {
+ return Collections.emptyList();
+ }
+
/**
* Setup management according to whether JMX is enabled or disabled.
*
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/StepListener.java
b/core/camel-api/src/main/java/org/apache/camel/spi/StepListener.java
new file mode 100644
index 000000000000..79c09c5ff68d
--- /dev/null
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/StepListener.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.spi;
+
+import org.apache.camel.Exchange;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * A listener that observes each call of the processor of a step of a route:
the call, its return or what it throws, and
+ * each call of its callback. A step is any node of a route with its node id
({@code to}, {@code process},
+ * {@code split}, ...), the same steps the {@link StepHook} is asked for. A
listener only observes: it must not change
+ * the exchange, and what it throws cannot change the routing. It is meant for
development and test tools that check the
+ * contracts of {@link org.apache.camel.AsyncProcessor} and {@link
org.apache.camel.AsyncCallback}.
+ * <p/>
+ * Listeners are installed with {@link
org.apache.camel.ExtendedCamelContext#addStepListener(StepListener)}. Unlike the
+ * {@link StepHook}, many can be installed. They are called in the order they
were added, and must not depend on that
+ * order.
+ * <p/>
+ * <b>Where it is called.</b> Whether a step gets the listeners is decided
when its route is created. When at least one
+ * listener is installed then, the channel of each step (the default channel
of Camel, not a channel of a custom
+ * {@link InternalProcessorFactory}) adds a small wrapper around the processor
of the step and the
+ * {@link InterceptStrategy} wrappers of the step, below the error handler of
the step. So the listeners see each call
+ * of that chain: each attempt, also each redelivery of the error handler, and
each call by the channel of a step
+ * without such an error handler. The {@link StepHook} is asked before the
wrapper is called, so a step that the hook
+ * fails is not a call the listeners see. A route that is created while no
listener is installed has no wrapper and is
+ * not seen until it is created again; add the listeners before the routes are
created. With no listener installed when
+ * a route is created, the route is built as without this SPI, and a step
costs nothing more. A step that has the
+ * wrapper reads the listeners installed at the start of each call, so a
listener added later is called by the routes
+ * that have the wrapper.
+ * <p/>
+ * With no {@link InterceptStrategy} on the step, the listeners see the call
of the processor of the step itself. A
+ * strategy can hide part of that call: {@code intercept()} calls the
processor later, after the intercept route, so the
+ * listeners do not see its own return value or what it throws; a strategy
that returns a plain
+ * {@link org.apache.camel.Processor} makes the call wait until the processor
is done, so a second call of the callback
+ * of the processor is not seen; and a strategy that catches what the
processor throws hides the throw.
+ * <p/>
+ * <b>The state of a call.</b> {@link #onStepCall(Exchange, String, String)}
returns a state for the call, and the other
+ * methods get it back. The state is held by the callback of the call, so it
lives as long as anything can still call
+ * that callback; it should therefore hold ids, names, threads or stacks,
never the exchange.
+ * <p/>
+ * <b>Threads.</b> The listeners are called on the thread of the event, by
many threads at the same time: they must be
+ * thread-safe and quick. Only {@link #onStepCall(Exchange, String, String)}
may read the exchange. The other methods
+ * can run on a thread that does not own the exchange, so they get only the
state.
+ * {@link #onStepCallback(Object, boolean)} can come before {@link
#onStepReturn(Object, boolean)}, on the same thread
+ * (the callback is called before {@code process} returns) or on another one.
+ * <p/>
+ * <b>Failures.</b> What a listener throws cannot change the routing. It is
caught and logged at WARN once per listener
+ * (then at DEBUG). The processor is called whatever {@link
#onStepCall(Exchange, String, String)} did, the callback of
+ * the call is called after the listeners, and what the processor throws is
thrown on unchanged. A listener whose
+ * {@link #onStepCall(Exchange, String, String)} threw gets no more events for
that call.
+ * <p/>
+ * A call keeps the listeners that were installed when it started, so a
listener that is removed meanwhile still gets
+ * the callback of a call it saw start.
+ * <p/>
+ * All methods have an empty default. Their names differ from {@link
StepHook#beforeStep(Exchange, String)}, so one
+ * class can implement both.
+ *
+ * @since 4.24
+ */
+public interface StepListener {
+
+ /**
+ * The processor of a step is about to be called (each attempt, also each
redelivery).
+ *
+ * @param exchange the exchange
+ * @param routeId the id of the route of the step
+ * @param nodeId the id of the step: the id of its node in the route,
such as {@code to1}, or the id set with
+ * {@code id("myStep")}
+ * @return the state of this call, passed to the other methods of
this listener for this call; can be
+ * {@code null}
+ */
+ default @Nullable Object onStepCall(Exchange exchange, String routeId,
String nodeId) {
+ return null;
+ }
+
+ /**
+ * {@code process(exchange, callback)} of the call returned.
+ *
+ * @param state the state returned by {@link #onStepCall(Exchange,
String, String)} for this call
+ * @param doneSync the value {@code process} returned
+ */
+ default void onStepReturn(@Nullable Object state, boolean doneSync) {
+ }
+
+ /**
+ * {@code process(exchange, callback)} of the call threw. The exception is
thrown on unchanged after the listeners.
+ *
+ * @param state the state returned by {@link #onStepCall(Exchange, String,
String)} for this call
+ * @param cause what {@code process} threw
+ */
+ default void onStepThrow(@Nullable Object state, Throwable cause) {
+ }
+
+ /**
+ * The callback of the call is called, before it continues the exchange.
It is called each time the callback is
+ * called, so a second call of the callback is seen too.
+ *
+ * @param state the state returned by {@link #onStepCall(Exchange,
String, String)} for this call
+ * @param doneSync the value the callback is called with
+ */
+ default void onStepCallback(@Nullable Object state, boolean doneSync) {
+ }
+
+}
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/Synchronization.java
b/core/camel-api/src/main/java/org/apache/camel/spi/Synchronization.java
index 22d5202a904d..5c288e3676d9 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/Synchronization.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/Synchronization.java
@@ -23,6 +23,26 @@ import org.jspecify.annotations.Nullable;
* Provides a hook for custom {@link org.apache.camel.Processor} or {@link
org.apache.camel.Component} instances to
* respond to completed or failed processing of an {@link Exchange} rather
like Spring's
* {@code org.springframework.transaction.support.TransactionSynchronization}
+ * <p/>
+ * <b>The commit of a consumer.</b> A consumer typically adds a
synchronization to the exchange before it routes it, to
+ * commit or roll back its message when the exchange is done (such as moving
or deleting a file). It moves into the unit
+ * of work that is created for the exchange. For such a commit the routing
engine keeps this contract:
+ * <ul>
+ * <li>Each time it is added, exactly one of {@link #onComplete(Exchange)} or
{@link #onFailure(Exchange)} is called,
+ * once (unless it is removed with {@link
UnitOfWork#removeSynchronization(Synchronization)}).</li>
+ * <li>It runs where it is held at the end: when the unit of work it was added
to is done, with the outcome of its
+ * exchange; or, when it was handed over, where it was handed over to. An
exchange that has no unit of work yet (such as
+ * a copy waiting in a {@code seda} queue) holds its commits itself, and they
move into the unit of work it gets, or
+ * back to the exchange that handed them over when the copy is not routed
(such as a copy that {@code seda} does not
+ * queue).</li>
+ * <li>While it is in a unit of work, it moves only by a handover through that
unit of work made by the exchange the
+ * unit of work was created for: {@link
org.apache.camel.ExchangeExtension#handoverCompletions(Exchange)} moves it to a
+ * copy that is routed on its own (such as an {@code InOnly} exchange sent to
{@code seda}), and it runs when the copy
+ * is done, with the outcome of the copy; {@link
org.apache.camel.ExchangeExtension#handoverCompletions()} takes it out
+ * to be run by someone else (such as the aggregator, the enrichers or {@code
onCompletion}). A
+ * {@link SynchronizationVetoable} can refuse to be handed over. The parts of
a split, multicast or recipient list with
+ * {@code shareUnitOfWork} use the unit of work of their parent, and do not
hand over the commit of the parent.</li>
+ * </ul>
*/
public interface Synchronization {
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultCamelContextExtension.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultCamelContextExtension.java
index 180bfcf79cff..7f31883855bc 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultCamelContextExtension.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultCamelContextExtension.java
@@ -17,6 +17,7 @@
package org.apache.camel.impl.engine;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -78,6 +79,7 @@ import org.apache.camel.spi.SendHook;
import org.apache.camel.spi.ShutdownStrategy;
import org.apache.camel.spi.StartupStepRecorder;
import org.apache.camel.spi.StepHook;
+import org.apache.camel.spi.StepListener;
import org.apache.camel.spi.StreamCachingStrategy;
import org.apache.camel.spi.Tracer;
import org.apache.camel.spi.TransformerRegistry;
@@ -89,6 +91,7 @@ import org.apache.camel.support.NormalizedUri;
import org.apache.camel.support.PluginHelper;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.support.startup.DefaultStartupStepRecorder;
+import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.StringHelper;
import org.apache.camel.util.URISupport;
import org.apache.camel.util.concurrent.ContextValue;
@@ -105,6 +108,8 @@ class DefaultCamelContextExtension implements
ExtendedCamelContext {
private final Map<String, FactoryFinder> factories = new
ConcurrentHashMap<>();
private final Map<String, FactoryFinder> bootstrapFactories = new
ConcurrentHashMap<>();
private final Set<LogListener> logListeners = new LinkedHashSet<>();
+ private final CopyOnWriteArrayList<StepListener> stepListeners = new
CopyOnWriteArrayList<>();
+ private final List<StepListener> stepListenersView =
Collections.unmodifiableList(stepListeners);
private final PluginManager pluginManager = new
DefaultContextPluginManager();
private final RouteController internalRouteController;
@@ -535,6 +540,23 @@ class DefaultCamelContextExtension implements
ExtendedCamelContext {
this.stepHook = stepHook;
}
+ @Override
+ public void addStepListener(StepListener listener) {
+ ObjectHelper.notNull(listener, "listener");
+ // a listener equal to one already installed is ignored
+ stepListeners.addIfAbsent(listener);
+ }
+
+ @Override
+ public void removeStepListener(StepListener listener) {
+ stepListeners.remove(listener);
+ }
+
+ @Override
+ public List<StepListener> getStepListeners() {
+ return stepListenersView;
+ }
+
@Override
public Set<LogListener> getLogListeners() {
return logListeners;
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
index 28f45d41c63a..e7a287ab9efb 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
@@ -41,6 +41,7 @@ import org.apache.camel.spi.InterceptStrategy;
import org.apache.camel.spi.InterceptableProcessor;
import org.apache.camel.spi.ManagementInterceptStrategy;
import org.apache.camel.spi.MessageHistoryFactory;
+import org.apache.camel.spi.StepListener;
import org.apache.camel.spi.Tracer;
import org.apache.camel.spi.WrapAwareProcessor;
import org.apache.camel.support.OrderedComparator;
@@ -265,6 +266,14 @@ public class DefaultChannel extends CamelInternalProcessor
implements Channel {
}
}
+ // the step listeners observe each call of what the strategies
returned, below the error handler of the step;
+ // with no listener installed there is no wrapper, so the step is
built as without listeners
+ List<StepListener> stepListeners =
camelContext.getCamelContextExtension().getStepListeners();
+ if (!stepListeners.isEmpty()) {
+ target = StepListenerProcessor.create(stepListeners,
route.getRouteId(), targetOutputDef.getId(),
+ nextProcessor, target);
+ }
+
if (route.isStreamCaching()) {
addAdvice(new
StreamCachingAdvice(camelContext.getStreamCachingStrategy()));
}
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/StepListenerProcessor.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/StepListenerProcessor.java
new file mode 100644
index 000000000000..ac99bed41a0d
--- /dev/null
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/StepListenerProcessor.java
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.impl.engine;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.WeakHashMap;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.spi.ErrorHandlerAware;
+import org.apache.camel.spi.StepListener;
+import org.apache.camel.support.processor.DelegateAsyncProcessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Calls the {@link StepListener}s around each call of the processor of a
step. {@link DefaultChannel} adds it after the
+ * {@link org.apache.camel.spi.InterceptStrategy} wrappers of the step, only
when a listener is installed when the route
+ * is created, so it is the output of the channel that the error handler of
the step calls for each attempt.
+ */
+class StepListenerProcessor extends DelegateAsyncProcessor {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(StepListenerProcessor.class);
+
+ private static final StepListener[] NO_LISTENERS = new StepListener[0];
+
+ // the state of a listener whose onStepCall threw: it gets no more events
for that call
+ private static final Object FAILED = new Object();
+
+ // the listeners that already failed once, so they are logged at WARN once
(weak keys: a listener is not kept)
+ private static final Map<StepListener, Boolean> FAILED_LISTENERS =
Collections.synchronizedMap(new WeakHashMap<>());
+
+ private final List<StepListener> listeners;
+ private final Processor step;
+ private final String routeId;
+ private final String nodeId;
+
+ StepListenerProcessor(List<StepListener> listeners, String routeId, String
nodeId, Processor step,
+ Processor processor) {
+ super(processor);
+ this.listeners = listeners;
+ this.routeId = routeId;
+ this.nodeId = nodeId;
+ this.step = step;
+ }
+
+ /**
+ * Creates the wrapper, which is also {@link ErrorHandlerAware} when what
it wraps is, so the error handler of the
+ * step is still given to the processor (multicast, recipient list, bean)
when the error handler of the step is
+ * created, as without the wrapper. (At the end of route creation, {@code
RouteReifier} gives these processors the
+ * error handler of the route anyway, as it finds them through {@link
org.apache.camel.Channel#getNextProcessor()}.)
+ */
+ static StepListenerProcessor create(
+ List<StepListener> listeners, String routeId, String nodeId,
Processor step, Processor processor) {
+ if (processor instanceof ErrorHandlerAware) {
+ return new ErrorHandlerAwareStepListenerProcessor(listeners,
routeId, nodeId, step, processor);
+ }
+ return new StepListenerProcessor(listeners, routeId, nodeId, step,
processor);
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ // the listeners when the call starts are kept for its callback
+ final StepListener[] called = listeners.toArray(NO_LISTENERS);
+ if (called.length == 0) {
+ return processor.process(exchange, callback);
+ }
+ final Object[] states = new Object[called.length];
+ for (int i = 0; i < called.length; i++) {
+ try {
+ states[i] = called[i].onStepCall(exchange, routeId, nodeId);
+ } catch (Throwable e) {
+ states[i] = FAILED;
+ failed(called[i], "onStepCall", e);
+ }
+ }
+ final StepCallback stepCallback = new StepCallback(called, states,
callback);
+ final boolean doneSync;
+ try {
+ doneSync = processor.process(exchange, stepCallback);
+ } catch (Throwable e) {
+ for (int i = 0; i < called.length; i++) {
+ if (states[i] != FAILED) {
+ try {
+ called[i].onStepThrow(states[i], e);
+ } catch (Throwable t) {
+ failed(called[i], "onStepThrow", t);
+ }
+ }
+ }
+ // what the processor threw is thrown on unchanged
+ throw e;
+ }
+ for (int i = 0; i < called.length; i++) {
+ if (states[i] != FAILED) {
+ try {
+ called[i].onStepReturn(states[i], doneSync);
+ } catch (Throwable t) {
+ failed(called[i], "onStepReturn", t);
+ }
+ }
+ }
+ return doneSync;
+ }
+
+ static void failed(StepListener listener, String method, Throwable cause) {
+ try {
+ if (FAILED_LISTENERS.put(listener, Boolean.TRUE) == null) {
+ LOG.warn("StepListener {} failed in {} (logged once for this
listener, then at DEBUG)", listener, method,
+ cause);
+ } else {
+ LOG.debug("StepListener {} failed in {}", listener, method,
cause);
+ }
+ } catch (Throwable t) {
+ // a listener cannot change the routing, also when logging it
fails (such as its toString)
+ }
+ }
+
+ @Override
+ public String toString() {
+ return step != null ? step.toString() : super.toString();
+ }
+
+ /**
+ * The callback of one call: calls the listeners, and then the callback of
the call, each time it is called.
+ */
+ private static final class StepCallback implements AsyncCallback {
+
+ private final StepListener[] called;
+ private final Object[] states;
+ private final AsyncCallback callback;
+
+ StepCallback(StepListener[] called, Object[] states, AsyncCallback
callback) {
+ this.called = called;
+ this.states = states;
+ this.callback = callback;
+ }
+
+ @Override
+ public void done(boolean doneSync) {
+ try {
+ for (int i = 0; i < called.length; i++) {
+ if (states[i] != FAILED) {
+ try {
+ called[i].onStepCallback(states[i], doneSync);
+ } catch (Throwable t) {
+ failed(called[i], "onStepCallback", t);
+ }
+ }
+ }
+ } finally {
+ callback.done(doneSync);
+ }
+ }
+ }
+
+ /**
+ * The wrapper of a processor that is {@link ErrorHandlerAware}.
+ */
+ private static final class ErrorHandlerAwareStepListenerProcessor extends
StepListenerProcessor
+ implements ErrorHandlerAware {
+
+ private final ErrorHandlerAware errorHandlerAware;
+
+ ErrorHandlerAwareStepListenerProcessor(List<StepListener> listeners,
String routeId, String nodeId,
+ Processor step, Processor
processor) {
+ super(listeners, routeId, nodeId, step, processor);
+ this.errorHandlerAware = (ErrorHandlerAware) processor;
+ }
+
+ @Override
+ public void setErrorHandler(Processor errorHandler) {
+ errorHandlerAware.setErrorHandler(errorHandler);
+ }
+
+ @Override
+ public Processor getErrorHandler() {
+ return errorHandlerAware.getErrorHandler();
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerInterceptStrategyTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerInterceptStrategyTest.java
new file mode 100644
index 000000000000..90687b498a1c
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerInterceptStrategyTest.java
@@ -0,0 +1,213 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Channel;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.NamedNode;
+import org.apache.camel.Processor;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.InterceptStrategy;
+import org.apache.camel.support.AsyncProcessorSupport;
+import org.apache.camel.support.processor.DelegateAsyncProcessor;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The {@link InterceptStrategy}s still get the processor of the step as their
target when step listeners are installed,
+ * and the listeners see the call of the chain of the strategies. With {@code
intercept()} and with a strategy that
+ * returns a plain {@link Processor}, the listeners see less of the call of
the processor, as documented.
+ */
+class StepListenerInterceptStrategyTest extends ContextTestSupport {
+
+ private final StepListenerRecorder recorder = new StepListenerRecorder();
+ // what the strategy got, per node id (the strategies are called on one
thread when the routes are created)
+ private final Map<String, Processor> targets =
Collections.synchronizedMap(new HashMap<>());
+ private final Map<String, Processor> nexts =
Collections.synchronizedMap(new HashMap<>());
+ private final ExecutorService worker = Executors.newSingleThreadExecutor(r
-> new Thread(r, "StepListenerWorker"));
+ private final CountDownLatch twiceDone = new CountDownLatch(2);
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getCamelContextExtension().addStepListener(recorder);
+ context.getCamelContextExtension().addInterceptStrategy(new
TypeCheckingStrategy());
+ return context;
+ }
+
+ @AfterEach
+ void stopWorker() {
+ worker.shutdownNow();
+ }
+
+ @Override
+ protected RoutesBuilder[] createRouteBuilders() {
+ return new RoutesBuilder[] { new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:typed").routeId("typed")
+ .to("mock:typed").id("typedTo");
+
+ // the processor calls its callback twice, on the worker thread
+ from("direct:twice").routeId("twice")
+ .errorHandler(noErrorHandler())
+ .process(new TwiceDone()).id("twiceStep")
+ .to("mock:twice").id("twiceTo");
+
+ from("direct:plain").routeId("plain")
+ .errorHandler(noErrorHandler())
+ .process(new TwiceDone()).id("plainStep")
+ .to("mock:plain").id("plainTo");
+ }
+ }, new RouteBuilder() {
+ @Override
+ public void configure() {
+ intercept().to("mock:intercepted").id("interceptTo");
+
+ from("direct:intercepted").routeId("intercepted")
+ .process(e ->
e.getMessage().setBody("Bye")).id("interceptedStep");
+ }
+ } };
+ }
+
+ @Test
+ void testStrategyGetsTheProcessorOfTheStep() throws Exception {
+ Channel channel =
StepListenerRecorder.channels(context.getRoute("typed")).get(0);
+ // as camel-opentelemetry checks the type of its target
+ assertInstanceOf(SendProcessor.class, targets.get("typedTo"));
+ assertSame(channel.getNextProcessor(), targets.get("typedTo"));
+ assertTrue(nexts.containsKey("typedTo"));
+ assertNull(nexts.get("typedTo"));
+
+ getMockEndpoint("mock:typed").expectedMessageCount(1);
+ template.sendBody("direct:typed", "Hello");
+ assertMockEndpointsSatisfied();
+
+ // the listener sees the call of the chain of the strategy
+ assertEquals(List.of("call typed/typedTo", "strategy typed/typedTo",
"callback typed/typedTo true",
+ "return typed/typedTo true"), recorder.of("typed"));
+ }
+
+ @Test
+ void testSecondCallbackIsSeenWithoutPlainProcessorStrategy() throws
Exception {
+ getMockEndpoint("mock:twice").expectedMessageCount(1);
+
+ template.sendBody("direct:twice", "Hello");
+
+ assertTrue(twiceDone.await(10, TimeUnit.SECONDS));
+ assertMockEndpointsSatisfied();
+ List<String> events = recorder.of("twice");
+ assertEquals(1, events.stream().filter(e -> e.equals("return
twice/twiceStep false")).count(), events.toString());
+ assertEquals(2, events.stream().filter(e -> e.equals("callback
twice/twiceStep false")).count(),
+ events.toString());
+ }
+
+ @Test
+ void testPlainProcessorStrategyHidesTheSecondCallback() throws Exception {
+ getMockEndpoint("mock:plain").expectedMessageCount(1);
+
+ template.sendBody("direct:plain", "Hello");
+
+ assertTrue(twiceDone.await(10, TimeUnit.SECONDS));
+ assertMockEndpointsSatisfied();
+ // the call waits in the await manager, so the listener sees one
synchronous call, and the second callback of
+ // the processor reaches the await manager, not the listener
+ List<String> events = recorder.of("plain");
+ assertEquals(List.of("call plain/plainStep", "callback plain/plainStep
true", "return plain/plainStep true"),
+ events.subList(0, 3));
+ assertEquals(1, events.stream().filter(e -> e.startsWith("callback
plain/plainStep ")).count(), events.toString());
+ }
+
+ @Test
+ void testInterceptHidesTheReturnValueOfTheProcessor() throws Exception {
+ getMockEndpoint("mock:intercepted").expectedMessageCount(1);
+
+ Object out = template.requestBody("direct:intercepted", "Hello");
+
+ assertEquals("Bye", out);
+ assertMockEndpointsSatisfied();
+ // the processor returns true, but the pipeline of intercept() returns
false and calls it later; the step of
+ // the intercept route is its own call
+ assertEquals(List.of(
+ "call intercepted/interceptedStep", "return
intercepted/interceptedStep false",
+ "call intercepted/interceptTo", "callback
intercepted/interceptTo true",
+ "return intercepted/interceptTo true", "callback
intercepted/interceptedStep false"),
+ recorder.of("intercepted"));
+ }
+
+ /**
+ * Calls its callback twice from the worker thread.
+ */
+ private final class TwiceDone extends AsyncProcessorSupport {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ worker.submit(() -> {
+ callback.done(false);
+ twiceDone.countDown();
+ callback.done(false);
+ twiceDone.countDown();
+ });
+ return false;
+ }
+ }
+
+ /**
+ * Wraps the steps whose target is a {@link SendProcessor} in the route
typed, as camel-opentelemetry checks the
+ * type of its target, and wraps the steps of the route plain in a plain
{@link Processor}.
+ */
+ private final class TypeCheckingStrategy implements InterceptStrategy {
+ @Override
+ public Processor wrapProcessorInInterceptors(
+ CamelContext context, NamedNode definition, Processor target,
Processor nextTarget) {
+ String id = definition.getId();
+ targets.put(id, target);
+ nexts.put(id, nextTarget);
+ if (id.startsWith("typed") && target instanceof SendProcessor) {
+ return new DelegateAsyncProcessor(target) {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback
callback) {
+ recorder.events.add("strategy typed/" + id);
+ return super.process(exchange, callback);
+ }
+ };
+ }
+ if ("plainStep".equals(id)) {
+ return target::process;
+ }
+ return target;
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerNotInstalledTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerNotInstalledTest.java
new file mode 100644
index 000000000000..70d42c32c64c
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerNotInstalledTest.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor;
+
+import java.util.List;
+
+import org.apache.camel.Channel;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Route;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.StepListener;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * With no {@link StepListener} installed when the routes are created, no
channel has the wrapper of the listeners: the
+ * error handler of each step (or the channel of a step without one) calls the
processor of the step itself, as before
+ * the listeners existed.
+ */
+class StepListenerNotInstalledTest extends ContextTestSupport {
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").routeId("start")
+ .setHeader("greeting", constant("Hello"))
+ .process(e -> e.getMessage().setBody("Bye"))
+ .doTry()
+ .to("mock:try")
+ .doCatch(IllegalArgumentException.class)
+ .to("mock:caught")
+ .doFinally()
+ .to("mock:finally")
+ .end()
+ .split(body())
+ .to("mock:split")
+ .end()
+ .multicast()
+ .to("mock:mc1")
+ .to("mock:mc2")
+ .end()
+ .to("mock:result");
+
+ from("direct:noErrorHandler").routeId("noErrorHandler")
+ .errorHandler(noErrorHandler())
+ .to("mock:noEh1")
+ .to("mock:noEh2");
+
+ from("direct:retry").routeId("retry")
+
.errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0))
+ .to("mock:retried");
+ }
+ };
+ }
+
+ @Test
+ void testNoWrapperWithoutListener() throws Exception {
+
assertTrue(context.getCamelContextExtension().getStepListeners().isEmpty());
+ int steps = 0;
+ for (Route route : context.getRoutes()) {
+ List<Channel> channels = StepListenerRecorder.channels(route);
+ for (Channel channel : channels) {
+ // the error handler (or the channel) calls the processor of
the step, not a wrapper
+ assertSame(channel.getNextProcessor(),
StepListenerRecorder.callOf(channel),
+ route.getRouteId() + " " + channel);
+ steps++;
+ }
+ }
+ // the steps of all the routes, also the nested ones
+ assertEquals(17, steps);
+
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+ template.sendBody("direct:start", "Hello");
+ assertMockEndpointsSatisfied();
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerRecorder.java
b/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerRecorder.java
new file mode 100644
index 000000000000..c668d0116691
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerRecorder.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.apache.camel.Channel;
+import org.apache.camel.Exchange;
+import org.apache.camel.Navigate;
+import org.apache.camel.Processor;
+import org.apache.camel.Route;
+import org.apache.camel.processor.errorhandler.ErrorHandlerSupport;
+import org.apache.camel.spi.StepListener;
+
+/**
+ * A {@link StepListener} for tests that records each event as a line {@code
"<event> <routeId>/<nodeId> [<value>]"},
+ * and the threads of the callbacks. Its state is the {@code
"<routeId>/<nodeId>"} of the call.
+ */
+class StepListenerRecorder implements StepListener {
+
+ final Queue<String> events = new ConcurrentLinkedQueue<>();
+ final Map<String, String> callbackThreads = new ConcurrentHashMap<>();
+ final Map<String, String> returnThreads = new ConcurrentHashMap<>();
+
+ @Override
+ public Object onStepCall(Exchange exchange, String routeId, String nodeId)
{
+ String state = routeId + "/" + nodeId;
+ events.add("call " + state);
+ return state;
+ }
+
+ @Override
+ public void onStepReturn(Object state, boolean doneSync) {
+ returnThreads.put((String) state, Thread.currentThread().getName());
+ events.add("return " + state + " " + doneSync);
+ }
+
+ @Override
+ public void onStepThrow(Object state, Throwable cause) {
+ events.add("throw " + state + " " + cause.getMessage());
+ }
+
+ @Override
+ public void onStepCallback(Object state, boolean doneSync) {
+ callbackThreads.put((String) state, Thread.currentThread().getName());
+ events.add("callback " + state + " " + doneSync);
+ }
+
+ /**
+ * The events of the steps of one route, in order.
+ */
+ List<String> of(String routeId) {
+ List<String> answer = new ArrayList<>();
+ for (String event : events) {
+ if (event.contains(" " + routeId + "/")) {
+ answer.add(event);
+ }
+ }
+ return answer;
+ }
+
+ /**
+ * The channels of all the steps of a route, also the nested ones.
+ */
+ static List<Channel> channels(Route route) {
+ List<Channel> answer = new ArrayList<>();
+ collect(route.navigate(), answer);
+ return answer;
+ }
+
+ private static void collect(Object node, List<Channel> answer) {
+ if (node instanceof Channel channel) {
+ answer.add(channel);
+ collect(channel.getNextProcessor(), answer);
+ } else if (node instanceof Navigate<?> navigate && navigate.hasNext())
{
+ for (Object child : navigate.next()) {
+ collect(child, answer);
+ }
+ }
+ }
+
+ /**
+ * What the error handler of the channel calls for each attempt, or the
channel itself when the step has no error
+ * handler.
+ */
+ static Processor callOf(Channel channel) {
+ Processor errorHandler = channel.getErrorHandler();
+ if (errorHandler instanceof ErrorHandlerSupport support) {
+ return support.getOutput();
+ }
+ return channel.getOutput();
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerTest.java
new file mode 100644
index 000000000000..a9f089bc1cf0
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/StepListenerTest.java
@@ -0,0 +1,539 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Channel;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Route;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.ErrorHandlerAware;
+import org.apache.camel.spi.StepHook;
+import org.apache.camel.spi.StepListener;
+import org.apache.camel.support.AsyncProcessorSupport;
+import org.apache.camel.support.processor.DelegateAsyncProcessor;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The {@link StepListener}s see each call of the processor of each step: each
attempt of the error handler, the steps
+ * without an error handler, the steps of doTry, doCatch and doFinally, a
multicast and its branches, and callbacks on
+ * another thread. They run with a {@link StepHook}, and a listener that fails
does not change the routing.
+ */
+class StepListenerTest extends ContextTestSupport {
+
+ private final StepListenerRecorder recorder = new StepListenerRecorder();
+ private final FailingListener failing = new FailingListener();
+ private final Map<String, AtomicInteger> hookAsked = new
ConcurrentHashMap<>();
+ private final ExecutorService worker = Executors.newSingleThreadExecutor(r
-> new Thread(r, "StepListenerWorker"));
+ // the async step calls its callback on the worker thread after the
listener saw process return
+ private final CountDownLatch asyncReturned = new CountDownLatch(1);
+ private final CountDownLatch gate = new CountDownLatch(1);
+ private final IllegalStateException forced = new
IllegalStateException("Forced");
+ private final AtomicReference<Object> caught = new AtomicReference<>();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ // the failing listener first, so its failures come before the
recorder is called
+ context.getCamelContextExtension().addStepListener(failing);
+ context.getCamelContextExtension().addStepListener(recorder);
+ context.getCamelContextExtension().setStepHook(this::beforeStep);
+ return context;
+ }
+
+ private long beforeStep(Exchange exchange, String nodeId) {
+ int n = hookAsked.computeIfAbsent(nodeId, k -> new
AtomicInteger()).incrementAndGet();
+ if ("hookedStep".equals(nodeId) && n == 1) {
+ exchange.setException(new IllegalArgumentException("Failed by the
step hook"));
+ }
+ return StepHook.PROCEED;
+ }
+
+ @AfterEach
+ void stopWorker() {
+ gate.countDown();
+ worker.shutdownNow();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:steps").routeId("steps")
+ .setHeader("greeting",
constant("Hello")).id("setHeader")
+ .process(e ->
e.getMessage().setBody("Bye")).id("process")
+ .to("mock:result").id("toResult");
+
+ AtomicInteger flakyCalls = new AtomicInteger();
+ from("direct:retry").routeId("retry")
+
.errorHandler(defaultErrorHandler().maximumRedeliveries(2).redeliveryDelay(0))
+ .process(e -> {
+ if (flakyCalls.incrementAndGet() <= 2) {
+ e.setException(new
IllegalArgumentException("Flaky " + flakyCalls.get()));
+ }
+ }).id("flaky")
+ .to("mock:retried").id("retried");
+
+ AtomicInteger throwingCalls = new AtomicInteger();
+ from("direct:throwOnce").routeId("throwOnce")
+
.errorHandler(defaultErrorHandler().maximumRedeliveries(1).redeliveryDelay(0))
+ .process(new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange,
AsyncCallback callback) {
+ if (throwingCalls.incrementAndGet() == 1) {
+ throw new IllegalStateException("Thrown
once");
+ }
+ callback.done(true);
+ return true;
+ }
+ }).id("throwsOnce")
+ .to("mock:afterThrowOnce").id("afterThrowOnce");
+
+ AtomicInteger branchCalls = new AtomicInteger();
+ from("direct:multicast").routeId("multicast")
+
.errorHandler(defaultErrorHandler().maximumRedeliveries(1).redeliveryDelay(0))
+ .multicast().id("multicastStep")
+ .process(e -> {
+ if (branchCalls.incrementAndGet() == 1) {
+ e.setException(new
IllegalArgumentException("Failed branch"));
+ }
+ }).id("failingBranch")
+ .to("mock:otherBranch").id("otherBranch")
+ .end()
+ .to("mock:afterMulticast").id("afterMulticast");
+
+ from("direct:noErrorHandler").routeId("noErrorHandler")
+ .errorHandler(noErrorHandler())
+ .to("mock:noEh1").id("noEh1")
+ .to("mock:noEh2").id("noEh2");
+
+ from("direct:try").routeId("try")
+ .doTry().id("doTry")
+ .process(e -> {
+ throw new IllegalArgumentException("Forced in
try");
+ }).id("inTry")
+ .doCatch(IllegalArgumentException.class).id("doCatch")
+ .to("mock:caught").id("inCatch")
+ .doFinally().id("doFinally")
+ .to("mock:finally").id("inFinally")
+ .end()
+ .to("mock:afterTry").id("afterTry");
+
+ from("direct:async").routeId("async")
+ .process(new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange,
AsyncCallback callback) {
+ worker.submit(() -> {
+ try {
+ asyncReturned.await(10,
TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ exchange.getMessage().setBody("Async");
+ callback.done(false);
+ });
+ return false;
+ }
+ }).id("asyncStep")
+ .to("mock:afterAsync").id("afterAsync");
+
+ from("direct:gated").routeId("gated")
+ .process(new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange,
AsyncCallback callback) {
+ worker.submit(() -> {
+ try {
+ gate.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ callback.done(false);
+ });
+ return false;
+ }
+ }).id("gatedStep")
+ .to("mock:gated").id("toGated");
+
+ from("direct:hooked").routeId("hooked")
+
.errorHandler(defaultErrorHandler().maximumRedeliveries(1).redeliveryDelay(0))
+ .to("mock:hooked").id("hookedStep");
+
+ from("direct:listenerFails").routeId("listenerFails")
+ .process(e ->
e.getMessage().setBody("A")).id("callFails")
+ .process(e ->
e.getMessage().setBody(e.getMessage().getBody(String.class) +
"B")).id("othersFail")
+ .to("mock:listenerFails").id("toListenerFails");
+
+ from("direct:throwUnchanged").routeId("throwUnchanged")
+ .onException(IllegalStateException.class).handled(true)
+ .process(e ->
caught.set(e.getProperty(Exchange.EXCEPTION_CAUGHT))).id("handler")
+ .end()
+ .process(new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange,
AsyncCallback callback) {
+ throw forced;
+ }
+ }).id("throwsUnchanged");
+ }
+ };
+ }
+
+ @Test
+ void testEveryStepIsSeen() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye");
+
+ template.sendBody("direct:steps", "Hello");
+
+ assertMockEndpointsSatisfied();
+ assertEquals(List.of(
+ "call steps/setHeader", "callback steps/setHeader true",
"return steps/setHeader true",
+ "call steps/process", "callback steps/process true", "return
steps/process true",
+ "call steps/toResult", "callback steps/toResult true", "return
steps/toResult true"),
+ recorder.of("steps"));
+ }
+
+ @Test
+ void testEachRedeliveryIsSeen() throws Exception {
+ getMockEndpoint("mock:retried").expectedMessageCount(1);
+
+ template.sendBody("direct:retry", "Hello");
+
+ assertMockEndpointsSatisfied();
+ List<String> flaky = new ArrayList<>(recorder.of("retry"));
+ flaky.removeIf(e -> !e.contains("/flaky"));
+ assertEquals(List.of(
+ "call retry/flaky", "callback retry/flaky true", "return
retry/flaky true",
+ "call retry/flaky", "callback retry/flaky true", "return
retry/flaky true",
+ "call retry/flaky", "callback retry/flaky true", "return
retry/flaky true"),
+ flaky);
+ }
+
+ @Test
+ void testProcessorThatThrowsIsSeenAndRedelivered() throws Exception {
+ getMockEndpoint("mock:afterThrowOnce").expectedMessageCount(1);
+
+ template.sendBody("direct:throwOnce", "Hello");
+
+ assertMockEndpointsSatisfied();
+ assertEquals(List.of(
+ "call throwOnce/throwsOnce", "throw throwOnce/throwsOnce
Thrown once",
+ "call throwOnce/throwsOnce", "callback throwOnce/throwsOnce
true", "return throwOnce/throwsOnce true",
+ "call throwOnce/afterThrowOnce", "callback
throwOnce/afterThrowOnce true",
+ "return throwOnce/afterThrowOnce true"),
+ recorder.of("throwOnce"));
+ }
+
+ @Test
+ void testErrorHandlerAwareStepAndItsBranches() throws Exception {
+ getMockEndpoint("mock:otherBranch").expectedMessageCount(1);
+ getMockEndpoint("mock:afterMulticast").expectedMessageCount(1);
+
+ template.sendBody("direct:multicast", "Hello");
+
+ assertMockEndpointsSatisfied();
+ List<String> events = recorder.of("multicast");
+ // the multicast once, and the failing branch once for each attempt of
the error handler that the multicast
+ // wraps its branches in
+ assertEquals(1, count(events, "call multicast/multicastStep"),
events.toString());
+ assertEquals(1, count(events, "callback multicast/multicastStep"),
events.toString());
+ assertEquals(1, count(events, "return multicast/multicastStep"),
events.toString());
+ assertEquals(2, count(events, "call multicast/failingBranch"),
events.toString());
+ assertEquals(2, count(events, "callback multicast/failingBranch"),
events.toString());
+ assertEquals(1, count(events, "call multicast/otherBranch"),
events.toString());
+ assertEquals(1, count(events, "callback multicast/otherBranch"),
events.toString());
+ assertEquals(1, count(events, "call multicast/afterMulticast"),
events.toString());
+ assertTrue(events.indexOf("call multicast/multicastStep") <
events.indexOf("call multicast/failingBranch"),
+ events.toString());
+ assertTrue(events.indexOf("call multicast/otherBranch") <
firstIndex(events, "callback multicast/multicastStep"),
+ events.toString());
+
+ // the wrapper of the multicast is ErrorHandlerAware, and passes the
error handler of the step through
+ Channel channel =
StepListenerRecorder.channels(context.getRoute("multicast")).stream()
+ .filter(c -> c.getNextProcessor() instanceof
MulticastProcessor)
+ .findFirst().orElseThrow();
+ MulticastProcessor multicast = (MulticastProcessor)
channel.getNextProcessor();
+ Processor call = StepListenerRecorder.callOf(channel);
+ assertEquals("ErrorHandlerAwareStepListenerProcessor",
call.getClass().getSimpleName());
+ ErrorHandlerAware wrapper = assertInstanceOf(ErrorHandlerAware.class,
call);
+ Processor errorHandler = multicast.getErrorHandler();
+ assertNotNull(errorHandler);
+ assertSame(errorHandler, wrapper.getErrorHandler());
+ Processor other = e -> {
+ };
+ wrapper.setErrorHandler(other);
+ assertSame(other, multicast.getErrorHandler());
+ wrapper.setErrorHandler(errorHandler);
+ assertSame(errorHandler, multicast.getErrorHandler());
+ }
+
+ /**
+ * The number of events of one kind for one step, such as {@code "callback
multicast/otherBranch"}, whatever their
+ * value.
+ */
+ private static long count(List<String> events, String event) {
+ return events.stream().filter(e -> isEvent(e, event)).count();
+ }
+
+ private static int firstIndex(List<String> events, String event) {
+ for (int i = 0; i < events.size(); i++) {
+ if (isEvent(events.get(i), event)) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ private static boolean isEvent(String recorded, String event) {
+ return recorded.equals(event) || recorded.startsWith(event + " ");
+ }
+
+ @Test
+ void testStepsWithoutErrorHandler() throws Exception {
+ getMockEndpoint("mock:noEh2").expectedMessageCount(1);
+
+ template.sendBody("direct:noErrorHandler", "Hello");
+
+ assertMockEndpointsSatisfied();
+ assertEquals(List.of(
+ "call noErrorHandler/noEh1", "callback noErrorHandler/noEh1
true", "return noErrorHandler/noEh1 true",
+ "call noErrorHandler/noEh2", "callback noErrorHandler/noEh2
true", "return noErrorHandler/noEh2 true"),
+ recorder.of("noErrorHandler"));
+ }
+
+ @Test
+ void testStepsOfDoTryDoCatchDoFinally() throws Exception {
+ getMockEndpoint("mock:caught").expectedMessageCount(1);
+ getMockEndpoint("mock:finally").expectedMessageCount(1);
+ getMockEndpoint("mock:afterTry").expectedMessageCount(1);
+
+ template.sendBody("direct:try", "Hello");
+
+ assertMockEndpointsSatisfied();
+ List<String> events = recorder.of("try");
+ // the doTry step itself, which the intercept strategies skip, and
each step inside it
+ for (String step : List.of("doTry", "inTry", "inCatch", "inFinally",
"afterTry")) {
+ assertEquals(1, events.stream().filter(e -> e.equals("call try/" +
step)).count(), step + " in " + events);
+ assertEquals(1, events.stream().filter(e -> e.startsWith("callback
try/" + step + " ")).count(),
+ step + " in " + events);
+ assertEquals(1, events.stream().filter(e -> e.startsWith("return
try/" + step + " ")).count(),
+ step + " in " + events);
+ }
+ assertTrue(events.indexOf("call try/doTry") < events.indexOf("call
try/inTry"), events.toString());
+ assertTrue(events.indexOf("call try/inTry") < events.indexOf("call
try/inCatch"), events.toString());
+ assertTrue(events.indexOf("call try/inCatch") < events.indexOf("call
try/inFinally"), events.toString());
+ assertTrue(events.indexOf("call try/inFinally") < events.indexOf("call
try/afterTry"), events.toString());
+ }
+
+ @Test
+ void testCallbackOnAnotherThread() throws Exception {
+ // the async step calls its callback once the listener saw process
return
+ StepListener onReturn = new StepListener() {
+ @Override
+ public void onStepReturn(Object state, boolean doneSync) {
+ asyncReturned.countDown();
+ }
+ };
+ context.getCamelContextExtension().addStepListener(onReturn);
+ getMockEndpoint("mock:afterAsync").expectedBodiesReceived("Async");
+
+ Object out = template.requestBody("direct:async", "Hello");
+
+ assertEquals("Async", out);
+ assertMockEndpointsSatisfied();
+ List<String> events = recorder.of("async");
+ assertEquals(List.of("call async/asyncStep", "return async/asyncStep
false", "callback async/asyncStep false"),
+ events.subList(0, 3));
+ assertEquals(Thread.currentThread().getName(),
recorder.returnThreads.get("async/asyncStep"));
+ assertEquals("StepListenerWorker",
recorder.callbackThreads.get("async/asyncStep"));
+ }
+
+ @Test
+ void testStepFailedByTheStepHookIsNotACall() throws Exception {
+ getMockEndpoint("mock:hooked").expectedMessageCount(1);
+
+ template.sendBody("direct:hooked", "Hello");
+
+ assertMockEndpointsSatisfied();
+ // the hook is asked for each attempt, and fails the first one before
the processor is called
+ assertEquals(2, hookAsked.get("hookedStep").get());
+ assertEquals(List.of("call hooked/hookedStep", "callback
hooked/hookedStep true", "return hooked/hookedStep true"),
+ recorder.of("hooked"));
+ }
+
+ @Test
+ void testFailingListenerDoesNotChangeTheRouting() throws Exception {
+ getMockEndpoint("mock:listenerFails").expectedBodiesReceived("AB");
+
+ template.sendBody("direct:listenerFails", "Hello");
+
+ assertMockEndpointsSatisfied();
+ // the other listener sees every event
+ assertEquals(List.of(
+ "call listenerFails/callFails", "callback
listenerFails/callFails true",
+ "return listenerFails/callFails true",
+ "call listenerFails/othersFail", "callback
listenerFails/othersFail true",
+ "return listenerFails/othersFail true",
+ "call listenerFails/toListenerFails", "callback
listenerFails/toListenerFails true",
+ "return listenerFails/toListenerFails true"),
+ recorder.of("listenerFails"));
+ // a listener whose onStepCall threw gets no more events for that call
+ assertEquals(List.of("call callFails", "call othersFail", "callback
othersFail", "return othersFail",
+ "call toListenerFails", "callback toListenerFails", "return
toListenerFails"),
+ new ArrayList<>(failing.events));
+ }
+
+ @Test
+ void testExceptionOfTheProcessorIsThrownOnUnchanged() throws Exception {
+ template.sendBody("direct:throwUnchanged", "Hello");
+
+ // the failing listener threw from onStepThrow, and the error handler
got the exception of the processor
+ assertSame(forced, caught.get());
+ assertTrue(failing.events.contains("throw throwsUnchanged"),
failing.events.toString());
+ assertTrue(recorder.of("throwUnchanged").contains("throw
throwUnchanged/throwsUnchanged Forced"),
+ recorder.of("throwUnchanged").toString());
+ }
+
+ @Test
+ void testRemovedListenerGetsTheCallbackOfACallItSawStart() throws
Exception {
+ StepListenerRecorder late = new StepListenerRecorder();
+ CountDownLatch lateCalled = new CountDownLatch(1);
+ // added after late, so late has recorded the return of the call when
the gate is opened
+ StepListener lateCall = new StepListener() {
+ @Override
+ public void onStepReturn(Object state, boolean doneSync) {
+ lateCalled.countDown();
+ }
+ };
+ context.getCamelContextExtension().addStepListener(late);
+ context.getCamelContextExtension().addStepListener(lateCall);
+ getMockEndpoint("mock:gated").expectedMessageCount(2);
+
+ Future<Object> first = template.asyncRequestBody("direct:gated",
"Hello");
+ assertTrue(lateCalled.await(10, TimeUnit.SECONDS));
+ context.getCamelContextExtension().removeStepListener(late);
+ context.getCamelContextExtension().removeStepListener(lateCall);
+
assertFalse(context.getCamelContextExtension().getStepListeners().contains(late));
+ gate.countDown();
+ first.get(10, TimeUnit.SECONDS);
+
+ // the step that started before the removal, and none after it
+ assertEquals(List.of("call gated/gatedStep", "return gated/gatedStep
false", "callback gated/gatedStep false"),
+ late.of("gated"));
+
+ template.requestBody("direct:gated", "Hello");
+ assertMockEndpointsSatisfied();
+ assertEquals(3, late.of("gated").size());
+ }
+
+ @Test
+ void testManyListenersInTheOrderTheyWereAdded() {
+ List<StepListener> listeners =
context.getCamelContextExtension().getStepListeners();
+ assertEquals(List.of(failing, recorder), listeners);
+ // the same instance twice is ignored
+ context.getCamelContextExtension().addStepListener(recorder);
+ assertEquals(List.of(failing, recorder),
context.getCamelContextExtension().getStepListeners());
+ assertThrows(UnsupportedOperationException.class, () ->
listeners.add(recorder));
+ }
+
+ @Test
+ void testTheWrapperIsBelowTheErrorHandlerAroundTheProcessor() {
+ Route route = context.getRoute("steps");
+ List<Channel> channels = StepListenerRecorder.channels(route);
+ assertEquals(3, channels.size());
+ for (Channel channel : channels) {
+ Processor step = channel.getNextProcessor();
+ Processor call = StepListenerRecorder.callOf(channel);
+ assertNotSame(step, call);
+ DelegateAsyncProcessor wrapper =
assertInstanceOf(DelegateAsyncProcessor.class, call);
+ assertEquals("StepListenerProcessor",
wrapper.getClass().getSimpleName());
+ assertSame(step, wrapper.getProcessor());
+ assertEquals(step.toString(), wrapper.toString());
+ }
+ // navigating the route still finds the processors of the steps
+ assertEquals(1, route.filter("toResult").size());
+ assertSame(channels.get(2).getNextProcessor(),
route.filter("toResult").get(0));
+ }
+
+ /**
+ * A listener that fails: in onStepCall at the step callFails, and in the
other methods at the steps othersFail and
+ * throwsUnchanged. It records the events it gets.
+ */
+ static final class FailingListener implements StepListener {
+
+ final Queue<String> events = new ConcurrentLinkedQueue<>();
+
+ @Override
+ public Object onStepCall(Exchange exchange, String routeId, String
nodeId) {
+ events.add("call " + nodeId);
+ if ("callFails".equals(nodeId)) {
+ throw new IllegalStateException("onStepCall failed");
+ }
+ return nodeId;
+ }
+
+ @Override
+ public void onStepReturn(Object state, boolean doneSync) {
+ events.add("return " + state);
+ failAt(state);
+ }
+
+ @Override
+ public void onStepThrow(Object state, Throwable cause) {
+ events.add("throw " + state);
+ failAt(state);
+ }
+
+ @Override
+ public void onStepCallback(Object state, boolean doneSync) {
+ events.add("callback " + state);
+ failAt(state);
+ }
+
+ private static void failAt(Object state) {
+ if ("othersFail".equals(state) || "throwsUnchanged".equals(state))
{
+ throw new IllegalStateException("Listener failed at " + state);
+ }
+ }
+ }
+}
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/ManagedStepListenerTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedStepListenerTest.java
new file mode 100644
index 000000000000..1dcbf901c992
--- /dev/null
+++
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedStepListenerTest.java
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.management;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Channel;
+import org.apache.camel.Exchange;
+import org.apache.camel.Navigate;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.processor.errorhandler.RedeliveryErrorHandler;
+import org.apache.camel.spi.BacklogDebugger;
+import org.apache.camel.spi.StepHook;
+import org.apache.camel.spi.StepListener;
+import org.apache.camel.support.processor.DelegateAsyncProcessor;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+import static
org.apache.camel.management.DefaultManagementObjectNameStrategy.TYPE_PROCESSOR;
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A {@link StepListener} runs together with a {@link StepHook} (as
camel-chaos uses), JMX, the tracer and the backlog
+ * debugger, and each still works: a step that the hook fails is not a call
the listener sees, JMX counts each call, the
+ * tracer traces and the debugger stops at its breakpoint before the listener
sees the step.
+ */
+@DisabledOnOs(OS.AIX)
+class ManagedStepListenerTest extends ManagementTestSupport {
+
+ private final Queue<String> events = new ConcurrentLinkedQueue<>();
+ private final AtomicInteger hookAsked = new AtomicInteger();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getCamelContextExtension().addStepListener(new StepListener() {
+ @Override
+ public Object onStepCall(Exchange exchange, String routeId, String
nodeId) {
+ events.add("call " + nodeId);
+ return nodeId;
+ }
+
+ @Override
+ public void onStepCallback(Object state, boolean doneSync) {
+ events.add("callback " + state);
+ }
+ });
+ context.getCamelContextExtension().setStepHook(this::beforeStep);
+ return context;
+ }
+
+ private long beforeStep(Exchange exchange, String nodeId) {
+ if ("flaky".equals(nodeId) && hookAsked.incrementAndGet() == 1) {
+ exchange.setException(new IllegalArgumentException("Failed by the
step hook"));
+ }
+ return StepHook.PROCEED;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.setTracing(true);
+ context.setDebugging(true);
+
+ from("direct:combo").routeId("combo")
+
.errorHandler(defaultErrorHandler().maximumRedeliveries(2).redeliveryDelay(0).logStackTrace(false))
+ .to("mock:flaky").id("flaky")
+ .to("mock:bar").id("bar")
+ .to("mock:result").id("result");
+ }
+ };
+ }
+
+ @Test
+ void testTogetherWithStepHookJmxTracerAndDebugger() throws Exception {
+ // with JMX and an error handler that can redeliver: error handler,
JMX instrumentation, the step hook
+ // processor, the listener wrapper, the processor
+ Channel channel = findChannel(context.getRoute("combo").navigate(),
"flaky");
+ assertNotNull(channel);
+ RedeliveryErrorHandler errorHandler =
assertInstanceOf(RedeliveryErrorHandler.class, channel.getErrorHandler());
+ DelegateAsyncProcessor instrumentation
+ = assertInstanceOf(DefaultInstrumentationProcessor.class,
errorHandler.getOutput());
+ DelegateAsyncProcessor hook =
assertInstanceOf(DelegateAsyncProcessor.class, instrumentation.getProcessor());
+ assertEquals("StepHookProcessor", hook.getClass().getSimpleName());
+ DelegateAsyncProcessor wrapper =
assertInstanceOf(DelegateAsyncProcessor.class, hook.getProcessor());
+ assertEquals("StepListenerProcessor",
wrapper.getClass().getSimpleName());
+ assertSame(channel.getNextProcessor(), wrapper.getProcessor());
+
+ BacklogDebugger debugger = context.hasService(BacklogDebugger.class);
+ assertNotNull(debugger);
+ debugger.enableDebugger();
+ debugger.addBreakpoint("bar");
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+
+ Future<Object> out = template.asyncRequestBody("direct:combo",
"Hello");
+
+ // the debugger stops before the step bar, so the listener has not
seen it yet
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
debugger.getSuspendedBreakpointNodeIds().contains("bar"));
+ assertFalse(events.contains("call bar"), events.toString());
+ debugger.resumeBreakpoint("bar");
+ assertEquals("Hello", out.get(10, TimeUnit.SECONDS));
+ assertMockEndpointsSatisfied();
+
+ // the hook failed the first attempt of flaky before its processor was
called, so the listener sees one call
+ assertEquals(2, hookAsked.get());
+ assertEquals(List.of("call flaky", "callback flaky", "call bar",
"callback bar", "call result",
+ "callback result"), new ArrayList<>(events));
+
+ // JMX counts the attempt that the hook failed as a failure of the step
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on = getCamelObjectName(TYPE_PROCESSOR, "flaky");
+ assertEquals(1L, mbeanServer.getAttribute(on, "ExchangesFailed"));
+ assertEquals(1L, mbeanServer.getAttribute(on, "ExchangesCompleted"));
+ on = getCamelObjectName(TYPE_PROCESSOR, "result");
+ assertEquals(1L, mbeanServer.getAttribute(on, "ExchangesCompleted"));
+
+ // the tracer traced the steps
+ assertTrue(context.getTracer().getTraceCounter() > 0);
+ }
+
+ private Channel findChannel(Object node, String nodeId) {
+ if (node instanceof Channel channel) {
+ Processor step = context.getRoute("combo").filter(nodeId).get(0);
+ if (channel.getNextProcessor() == step) {
+ return channel;
+ }
+ } else if (node instanceof Navigate<?> navigate && navigate.hasNext())
{
+ for (Object child : navigate.next()) {
+ Channel found = findChannel(child, nodeId);
+ if (found != null) {
+ return found;
+ }
+ }
+ }
+ return null;
+ }
+}
diff --git a/design/exchange-sanitizer.adoc b/design/exchange-sanitizer.adoc
index 318c57289f64..660871616c78 100644
--- a/design/exchange-sanitizer.adoc
+++ b/design/exchange-sanitizer.adoc
@@ -424,7 +424,7 @@ target, Predicate<Synchronization> filter)`), as today only
`DefaultUnitOfWork`
Rules for listeners, in the Javadoc:
* Many can be installed, each kind in a copy-on-write array on the context
extension (the getter returns a cached
- unmodifiable view, so a call site checks `isEmpty()` without copying; the
same instance twice is ignored). They are
+ unmodifiable view, so a call site checks `isEmpty()` without copying; an
equal listener is ignored). They are
called in the order they were added, on the thread of the event, by many
threads at once: they must be thread-safe
and quick (the R3 pauses are a deliberate exception of the sanitizer), and
must not depend on the order.
* A call of a step reads the listeners when it starts and keeps them for its
callback, so a listener removed meanwhile
