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 77bc9ccf0eaa CAMEL-26742: camel-core - on completion listener SPI for
the exchange sanitizer (#29436)
77bc9ccf0eaa is described below
commit 77bc9ccf0eaa6fe940b477ae8371b68636d15e32
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Nov 9 15:04:36 2026 +0530
CAMEL-26742: camel-core - on completion listener SPI for the exchange
sanitizer (#29436)
* CAMEL-26742: camel-core - on completion listener SPI for the exchange
sanitizer
* CAMEL-26742: camel-api - rename the handover method that returns what
moved to handoverSynchronizationWithResult
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/ExtendedCamelContext.java | 36 ++
.../org/apache/camel/spi/OnCompletionListener.java | 143 ++++++
.../main/java/org/apache/camel/spi/UnitOfWork.java | 26 ++
.../impl/engine/DefaultCamelContextExtension.java | 20 +
.../camel/impl/engine/DefaultUnitOfWork.java | 35 +-
.../OnCompletionListenerCustomUnitOfWorkTest.java | 111 +++++
.../camel/processor/OnCompletionListenerTest.java | 492 +++++++++++++++++++++
.../org/apache/camel/support/AbstractExchange.java | 45 +-
.../camel/support/OnCompletionListenerHelper.java | 145 ++++++
.../org/apache/camel/support/UnitOfWorkHelper.java | 12 +-
design/exchange-sanitizer.adoc | 20 +-
11 files changed, 1071 insertions(+), 14 deletions(-)
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 c7c3bd577c07..96a4adb29f47 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
@@ -38,6 +38,7 @@ import org.apache.camel.spi.LifecycleStrategy;
import org.apache.camel.spi.LogListener;
import org.apache.camel.spi.ManagementMBeanAssembler;
import org.apache.camel.spi.NormalizedEndpointUri;
+import org.apache.camel.spi.OnCompletionListener;
import org.apache.camel.spi.ProcessorExchangeFactory;
import org.apache.camel.spi.ReactiveExecutor;
import org.apache.camel.spi.Registry;
@@ -602,6 +603,41 @@ public interface ExtendedCamelContext {
return Collections.emptyList();
}
+ /**
+ * Adds an {@link OnCompletionListener} that observes where the on
completions of the exchanges are: when they move
+ * into a unit of work, are added to or removed from one, are handed over
or taken out, and when they run. Many
+ * listeners can be installed; adding a listener that is already installed
(equal to one installed) is ignored. The
+ * listeners are read at each event, so a listener added while the routes
run is called from the next event on. 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 addOnCompletionListener(OnCompletionListener listener) {
+ throw new UnsupportedOperationException("This CamelContext does not
support an OnCompletionListener");
+ }
+
+ /**
+ * Removes an {@link OnCompletionListener} (one equal to it). It is no
longer called from the next event on.
+ *
+ * @param listener the listener
+ * @since 4.24
+ */
+ default void removeOnCompletionListener(OnCompletionListener listener) {
+ // no listener can be installed, so there is nothing to remove
+ }
+
+ /**
+ * Gets the {@link OnCompletionListener}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<OnCompletionListener> getOnCompletionListeners() {
+ 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/OnCompletionListener.java
b/core/camel-api/src/main/java/org/apache/camel/spi/OnCompletionListener.java
new file mode 100644
index 000000000000..d0e5eee508e2
--- /dev/null
+++
b/core/camel-api/src/main/java/org/apache/camel/spi/OnCompletionListener.java
@@ -0,0 +1,143 @@
+/*
+ * 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 java.util.List;
+
+import org.apache.camel.Exchange;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * A listener that observes where the on completions ({@link
Synchronization}s) of the exchanges are: when an exchange
+ * gets a unit of work and its on completions move into it, when an on
completion is added to or removed from a unit of
+ * work, handed over to another exchange or taken out to be run by someone
else, and when it runs. A listener only
+ * observes: it must not change the exchanges, the units of work or the lists
it is given, and what it throws cannot
+ * change the routing. It is meant for development and test tools that check
the contract of the commit of a consumer,
+ * described on {@link Synchronization}.
+ * <p/>
+ * Listeners are installed with
+ * {@link
org.apache.camel.ExtendedCamelContext#addOnCompletionListener(OnCompletionListener)}.
Many can be installed.
+ * They are called in the order they were added, and must not depend on that
order. Unlike the {@link StepListener},
+ * they are read at each event, so a listener added or removed while the
routes run is called, or no longer called, from
+ * the next event on. With no listener installed, each of these places only
checks that the list of listeners is empty.
+ * <p/>
+ * <b>Where it is called.</b>
+ * <ul>
+ * <li>{@link #onUnitOfWork(Exchange, UnitOfWork, List)} when an exchange is
given a unit of work (also one it shares
+ * with a parent exchange), before the on completions it held until then move
into it.</li>
+ * <li>{@link #onCompletionAdded(UnitOfWork, Synchronization)} when an on
completion is added to a unit of work of Camel
+ * ({@link UnitOfWork#addSynchronization(Synchronization)}), also as part of
such a move or of a handover. The adds of a
+ * move come after the {@link #onUnitOfWork(Exchange, UnitOfWork, List)} of
the move, and the adds of a handover before
+ * the {@link #onCompletionHandover(Exchange, UnitOfWork, Exchange, List)} of
the handover, on the same thread, so a
+ * listener can tell them apart. Adds to an exchange that has no unit of work
yet are not reported: they show up in the
+ * move.</li>
+ * <li>{@link #onCompletionRemoved(UnitOfWork, Synchronization)} after an on
completion is removed from a unit of work
+ * of Camel ({@link UnitOfWork#removeSynchronization(Synchronization)}), so it
does not run.</li>
+ * <li>{@link #onCompletionHandover(Exchange, UnitOfWork, Exchange, List)}
after on completions are handed over from one
+ * exchange to another ({@link
org.apache.camel.ExchangeExtension#handoverCompletions(Exchange)}), such as to
a copy
+ * that is routed on its own, or back from a copy that is not routed.</li>
+ * <li>{@link #onCompletionTaken(Exchange, List)} after on completions are
taken out of an exchange as a list
+ * ({@link org.apache.camel.ExchangeExtension#handoverCompletions()}), to be
run by someone else.</li>
+ * <li>{@link #onCompletionRun(Exchange, Synchronization, boolean)} before an
on completion runs (when a unit of work is
+ * done, or when a list of on completions that was taken out is run).</li>
+ * </ul>
+ * Exchange created, completed and failed are events of the {@link
EventNotifier}, so they are not in this listener.
+ * <p/>
+ * <b>Threads.</b> The listeners are called on the thread that makes the
change, by many threads at the same time: they
+ * must be thread-safe and quick. The order above holds for the events of one
thread; events of different threads (such
+ * as the parts of a parallel split that share a unit of work) are not ordered
with each other. A listener may read the
+ * exchanges it is given on that thread. It should not keep the exchanges or
the on completions, so these can be
+ * released when the exchange is done. The lists it is given are unmodifiable,
and valid during the call only.
+ * <p/>
+ * An exchange that is created as a copy of another ({@link Exchange#copy()})
starts with the unit of work of that
+ * exchange, without {@link #onUnitOfWork(Exchange, UnitOfWork, List)}.
+ * <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), and the change or the run of the on completion happens as
without the listener.
+ * <p/>
+ * All methods have an empty default.
+ *
+ * @since 4.24
+ */
+public interface OnCompletionListener {
+
+ /**
+ * An exchange is given a unit of work. The on completions it held until
then move into it after this call.
+ *
+ * @param exchange the exchange
+ * @param unitOfWork the unit of work
+ * @param moved the on completions that move into the unit of work
(empty if none)
+ */
+ default void onUnitOfWork(Exchange exchange, UnitOfWork unitOfWork,
List<Synchronization> moved) {
+ }
+
+ /**
+ * An on completion is added to a unit of work.
+ *
+ * @param unitOfWork the unit of work
+ * @param onCompletion the on completion
+ */
+ default void onCompletionAdded(UnitOfWork unitOfWork, Synchronization
onCompletion) {
+ }
+
+ /**
+ * An on completion was removed from a unit of work, so it does not run.
Called only when it was there.
+ *
+ * @param unitOfWork the unit of work
+ * @param onCompletion the on completion
+ */
+ default void onCompletionRemoved(UnitOfWork unitOfWork, Synchronization
onCompletion) {
+ }
+
+ /**
+ * On completions were handed over from one exchange (and its unit of
work, if it has one) to another exchange.
+ * Called only when on completions moved, or when the unit of work does
not tell what moved.
+ *
+ * @param from the exchange that handed them over
+ * @param fromUnitOfWork the unit of work of that exchange they were
handed over from, or {@code null} when the
+ * exchange held them itself (it had no unit of work)
+ * @param to the exchange they were handed over to
+ * @param moved the on completions that moved (not empty), or
{@code null} when the unit of work does not
+ * tell (a custom {@link UnitOfWork} that does not
implement
+ * {@link
UnitOfWork#handoverSynchronizationWithResult(Exchange,
java.util.function.Predicate)})
+ */
+ default void onCompletionHandover(
+ Exchange from, @Nullable UnitOfWork fromUnitOfWork, Exchange to,
@Nullable List<Synchronization> moved) {
+ }
+
+ /**
+ * On completions were taken out of an exchange as a list, to be run by
someone else.
+ *
+ * @param from the exchange they were taken out of
+ * @param taken the on completions (not empty)
+ */
+ default void onCompletionTaken(Exchange from, List<Synchronization> taken)
{
+ }
+
+ /**
+ * An on completion is about to run {@link
Synchronization#onComplete(Exchange)} or
+ * {@link Synchronization#onFailure(Exchange)}.
+ *
+ * @param exchange the exchange it runs with
+ * @param onCompletion the on completion
+ * @param failure {@code true} if it runs {@link
Synchronization#onFailure(Exchange)}, {@code false} for
+ * {@link Synchronization#onComplete(Exchange)}
+ */
+ default void onCompletionRun(Exchange exchange, Synchronization
onCompletion, boolean failure) {
+ }
+
+}
diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/UnitOfWork.java
b/core/camel-api/src/main/java/org/apache/camel/spi/UnitOfWork.java
index 1204c1c71609..eb79d0400844 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/UnitOfWork.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/UnitOfWork.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.spi;
+import java.util.List;
import java.util.function.Predicate;
import org.apache.camel.AsyncCallback;
@@ -124,6 +125,31 @@ public interface UnitOfWork {
*/
void handoverSynchronization(Exchange target, @Nullable
Predicate<Synchronization> filter);
+ /**
+ * Handover the registered synchronizations to the target {@link
org.apache.camel.Exchange}, as
+ * {@link #handoverSynchronization(Exchange, Predicate)} does, and tells
which ones were handed over (a
+ * {@link SynchronizationVetoable} can refuse to be handed over). Camel
calls it instead of
+ * {@link #handoverSynchronization(Exchange, Predicate)} when an {@link
OnCompletionListener} is installed.
+ * <p/>
+ * The default hands over with {@link #handoverSynchronization(Exchange)}
or
+ * {@link #handoverSynchronization(Exchange, Predicate)}, as Camel does
without a listener, and returns
+ * {@code null}, as it does not know what was handed over; so a custom
{@link UnitOfWork} hands over as before.
+ *
+ * @param target the target exchange
+ * @param filter optional filter to only handover if filter returns
<tt>true</tt>
+ * @return the synchronizations that were handed over (empty if
none), or {@code null} if not known
+ * @since 4.24
+ */
+ default @Nullable List<Synchronization> handoverSynchronizationWithResult(
+ Exchange target, @Nullable Predicate<Synchronization> filter) {
+ if (filter == null) {
+ handoverSynchronization(target);
+ } else {
+ handoverSynchronization(target, filter);
+ }
+ return null;
+ }
+
/**
* Invoked when this unit of work has been completed, whether it has
failed or completed
*
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 25fafec3d447..01a79fd3590f 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
@@ -66,6 +66,7 @@ import org.apache.camel.spi.ManagementStrategyFactory;
import org.apache.camel.spi.MessageHistoryFactory;
import org.apache.camel.spi.MessageSizeStrategy;
import org.apache.camel.spi.NormalizedEndpointUri;
+import org.apache.camel.spi.OnCompletionListener;
import org.apache.camel.spi.PluginManager;
import org.apache.camel.spi.ProcessorExchangeFactory;
import org.apache.camel.spi.PropertiesComponent;
@@ -113,6 +114,8 @@ class DefaultCamelContextExtension implements
ExtendedCamelContext {
private final Set<LogListener> logListeners = new LinkedHashSet<>();
private final CopyOnWriteArrayList<StepListener> stepListeners = new
CopyOnWriteArrayList<>();
private final List<StepListener> stepListenersView =
Collections.unmodifiableList(stepListeners);
+ private final CopyOnWriteArrayList<OnCompletionListener>
onCompletionListeners = new CopyOnWriteArrayList<>();
+ private final List<OnCompletionListener> onCompletionListenersView =
Collections.unmodifiableList(onCompletionListeners);
private final PluginManager pluginManager = new
DefaultContextPluginManager();
private final RouteController internalRouteController;
@@ -604,6 +607,23 @@ class DefaultCamelContextExtension implements
ExtendedCamelContext {
return stepListenersView;
}
+ @Override
+ public void addOnCompletionListener(OnCompletionListener listener) {
+ ObjectHelper.notNull(listener, "listener");
+ // a listener equal to one already installed is ignored
+ onCompletionListeners.addIfAbsent(listener);
+ }
+
+ @Override
+ public void removeOnCompletionListener(OnCompletionListener listener) {
+ onCompletionListeners.remove(listener);
+ }
+
+ @Override
+ public List<OnCompletionListener> getOnCompletionListeners() {
+ return onCompletionListenersView;
+ }
+
@Override
public Set<LogListener> getLogListeners() {
return logListeners;
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java
index 87a72d721587..6579055ffc66 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java
@@ -35,11 +35,13 @@ import org.apache.camel.Processor;
import org.apache.camel.Route;
import org.apache.camel.StreamCache;
import org.apache.camel.spi.InflightRepository;
+import org.apache.camel.spi.OnCompletionListener;
import org.apache.camel.spi.StreamCachingStrategy;
import org.apache.camel.spi.Synchronization;
import org.apache.camel.spi.SynchronizationVetoable;
import org.apache.camel.spi.UnitOfWork;
import org.apache.camel.support.EventHelper;
+import org.apache.camel.support.OnCompletionListenerHelper;
import org.apache.camel.support.UnitOfWorkHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -200,18 +202,29 @@ public class DefaultUnitOfWork implements UnitOfWork {
} finally {
lock.unlock();
}
+ final List<OnCompletionListener> listeners =
OnCompletionListenerHelper.getListeners(context);
+ if (!listeners.isEmpty()) {
+ OnCompletionListenerHelper.onCompletionAdded(listeners, this,
synchronization);
+ }
}
@Override
public void removeSynchronization(Synchronization synchronization) {
+ boolean removed = false;
lock.lock();
try {
if (synchronizations != null) {
- synchronizations.remove(synchronization);
+ removed = synchronizations.remove(synchronization);
}
} finally {
lock.unlock();
}
+ if (removed) {
+ final List<OnCompletionListener> listeners =
OnCompletionListenerHelper.getListeners(context);
+ if (!listeners.isEmpty()) {
+ OnCompletionListenerHelper.onCompletionRemoved(listeners,
this, synchronization);
+ }
+ }
}
@Override
@@ -231,6 +244,23 @@ public class DefaultUnitOfWork implements UnitOfWork {
@Override
public void handoverSynchronization(Exchange target,
Predicate<Synchronization> filter) {
+ doHandoverSynchronization(target, filter, null);
+ }
+
+ /**
+ * Hands over as {@link #handoverSynchronization(Exchange, Predicate)}
does, and returns what was handed over. A
+ * subclass that changes how the synchronizations are handed over should
override this method too, as Camel calls it
+ * instead of the others when an {@link OnCompletionListener} is installed.
+ */
+ @Override
+ public List<Synchronization> handoverSynchronizationWithResult(Exchange
target, Predicate<Synchronization> filter) {
+ final List<Synchronization> moved = new ArrayList<>(2);
+ doHandoverSynchronization(target, filter, moved);
+ return moved;
+ }
+
+ private void doHandoverSynchronization(
+ Exchange target, Predicate<Synchronization> filter,
List<Synchronization> moved) {
// the parts of a parallel Split, Multicast or Recipient List with
shareUnitOfWork use this unit of work at the
// same time, so work on a copy, and take each synchronization out
under the lock before handing it over, so it
// is handed over only once (and the target is not called while
holding the lock)
@@ -252,6 +282,9 @@ public class DefaultUnitOfWork implements UnitOfWork {
if (takeSynchronization(synchronization)) {
log.trace("Handover synchronization {} to: {}",
synchronization, target);
target.getExchangeExtension().addOnCompletion(synchronization);
+ if (moved != null) {
+ moved.add(synchronization);
+ }
// Allow the synchronization to do housekeeping before
transfer
if (veto != null) {
veto.beforeHandover(target);
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionListenerCustomUnitOfWorkTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionListenerCustomUnitOfWorkTest.java
new file mode 100644
index 000000000000..97159740fbbd
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionListenerCustomUnitOfWorkTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.lang.reflect.InvocationHandler;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.util.List;
+import java.util.function.Predicate;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.engine.DefaultUnitOfWork;
+import org.apache.camel.processor.OnCompletionListenerTest.Commit;
+import org.apache.camel.processor.OnCompletionListenerTest.Event;
+import org.apache.camel.processor.OnCompletionListenerTest.Recorder;
+import org.apache.camel.processor.OnCompletionListenerTest.Type;
+import org.apache.camel.spi.UnitOfWork;
+import org.apache.camel.spi.UnitOfWorkFactory;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * With a custom {@link UnitOfWork} that does not implement
+ * {@link UnitOfWork#handoverSynchronizationWithResult(Exchange, Predicate)},
the on completions are handed over as
+ * without a listener, and the listener is told that it is not known what
moved.
+ */
+class OnCompletionListenerCustomUnitOfWorkTest extends ContextTestSupport {
+
+ private final Recorder recorder = new Recorder();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+
context.getCamelContextExtension().addContextPlugin(UnitOfWorkFactory.class,
CustomUnitOfWork::create);
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").to("seda:queue");
+ from("seda:queue").to("mock:queue");
+ }
+ };
+ }
+
+ @Test
+ void testHandoverThatDoesNotTellWhatMoved() throws Exception {
+ context.getCamelContextExtension().addOnCompletionListener(recorder);
+ Commit commit = new Commit("commit");
+
+ Exchange exchange = createExchangeWithBody("A");
+ exchange.getExchangeExtension().addOnCompletion(commit);
+ template.send("direct:start", exchange);
+ commit.awaitRun();
+
+ Event handover = recorder.events.stream()
+ .filter(e -> e.type() == Type.HANDOVER && e.exchange() ==
exchange)
+ .findFirst().orElseThrow();
+ assertNull(handover.list());
+ assertEquals(1, commit.runs.get());
+ // it ran with the copy, as it was handed over
+ Event run = recorder.events.stream()
+ .filter(e -> e.type() == Type.RUN && e.onCompletion() ==
commit)
+ .findFirst().orElseThrow();
+ assertEquals(List.of(run.exchange()), List.of(handover.to()));
+ }
+
+ /**
+ * A unit of work that delegates to the default one, but uses the default
method of the interface to hand over.
+ */
+ private static final class CustomUnitOfWork {
+
+ static UnitOfWork create(Exchange exchange) {
+ DefaultUnitOfWork delegate = new DefaultUnitOfWork(exchange);
+ InvocationHandler handler = (proxy, method, args) -> {
+ if (method.isDefault() &&
"handoverSynchronizationWithResult".equals(method.getName())) {
+ return InvocationHandler.invokeDefault(proxy, method,
args);
+ }
+ try {
+ return method.invoke(delegate, args);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ };
+ return (UnitOfWork)
Proxy.newProxyInstance(CustomUnitOfWork.class.getClassLoader(),
+ new Class<?>[] { UnitOfWork.class }, handler);
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionListenerTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionListenerTest.java
new file mode 100644
index 000000000000..633dd403d568
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/OnCompletionListenerTest.java
@@ -0,0 +1,492 @@
+/*
+ * 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.lang.reflect.InvocationHandler;
+import java.lang.reflect.Proxy;
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.OnCompletionListener;
+import org.apache.camel.spi.Synchronization;
+import org.apache.camel.spi.SynchronizationVetoable;
+import org.apache.camel.spi.UnitOfWork;
+import org.apache.camel.support.UnitOfWorkHelper;
+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.assertNull;
+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 OnCompletionListener} sees each move, add, handover, take and
run of an on completion, on the paths that
+ * Camel moves them today: a commit handed over to a copy sent to seda, the
copy that seda does not queue giving it back
+ * (CAMEL-25123), and the parts of a split with shareUnitOfWork, which hand
over only their own on completions
+ * (CAMEL-26576). With no listener, nothing changes.
+ */
+class OnCompletionListenerTest extends ContextTestSupport {
+
+ private static final long WAIT_SECONDS = 10;
+
+ private final Recorder recorder = new Recorder();
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").to("seda:queue");
+ from("seda:queue").to("mock:queue");
+
+ // no consumer, so the first message fills the queue and the
next one is discarded
+
from("direct:full").to("seda:full?size=1&discardWhenFull=true");
+
+ from("direct:share")
+ .split(body().tokenize(",")).shareUnitOfWork()
+ .process(e ->
e.getExchangeExtension().addOnCompletion((Commit) e.getProperty("own")))
+ .to("seda:parts")
+ .end();
+ from("seda:parts").to("mock:parts");
+
+ from("direct:remove")
+ .process(e ->
e.getUnitOfWork().addSynchronization((Commit) e.getProperty("removed")))
+ .process(e ->
e.getUnitOfWork().removeSynchronization((Commit) e.getProperty("removed")))
+ // it is no longer there, so this one is not reported
+ .process(e ->
e.getUnitOfWork().removeSynchronization((Commit) e.getProperty("removed")));
+ }
+ };
+ }
+
+ @Test
+ void testHandoverToACopySentToSeda() throws Exception {
+ context.getCamelContextExtension().addOnCompletionListener(recorder);
+ Commit commit = new Commit("commit");
+
+ Exchange exchange = createExchangeWithBody("A");
+ exchange.getExchangeExtension().addOnCompletion(commit);
+ template.send("direct:start", exchange);
+ commit.awaitRun();
+
+ assertEquals(List.of(
+ "unitOfWork e1 u1 [commit]",
+ "added u1 commit",
+ "handover e1 u1 -> e2 [commit]",
+ "unitOfWork e2 u2 [commit]",
+ "added u2 commit",
+ "run e2 commit complete"),
+ recorder.lines(commit));
+ assertSame(exchange, recorder.events.peek().exchange);
+ assertEquals(1, commit.runs.get());
+ }
+
+ @Test
+ void testCopyNotQueuedGivesItBack() throws Exception {
+ context.getCamelContextExtension().addOnCompletionListener(recorder);
+ Commit queued = new Commit("queued");
+ Commit discarded = new Commit("discarded");
+
+ Exchange first = createExchangeWithBody("A");
+ first.getExchangeExtension().addOnCompletion(queued);
+ template.send("direct:full", first);
+ Exchange second = createExchangeWithBody("B");
+ second.getExchangeExtension().addOnCompletion(discarded);
+ template.send("direct:full", second);
+ discarded.awaitRun();
+
+ // the copy has no unit of work, so it gives the commit back from its
own list, and the exchange runs it
+ assertEquals(List.of(
+ "unitOfWork e1 u1 [discarded]",
+ "added u1 discarded",
+ "handover e1 u1 -> e2 [discarded]",
+ "added u1 discarded",
+ "handover e2 - -> e1 [discarded]",
+ "run e1 discarded complete"),
+ recorder.lines(discarded));
+ assertEquals(1, discarded.runs.get());
+ // the queued copy holds the other commit, which has not run
+ assertEquals(List.of(
+ "unitOfWork e1 u1 [queued]",
+ "added u1 queued",
+ "handover e1 u1 -> e2 [queued]"),
+ recorder.lines(queued));
+ assertEquals(0, queued.runs.get());
+ }
+
+ @Test
+ void testSharedUnitOfWorkHandsOverOnlyThePartsOwn() throws Exception {
+ context.getCamelContextExtension().addOnCompletionListener(recorder);
+ Commit commit = new Commit("commit");
+ Commit own = new Commit("own");
+
+ Exchange exchange = createExchangeWithBody("A");
+ exchange.setProperty("own", own);
+ exchange.getExchangeExtension().addOnCompletion(commit);
+ template.send("direct:share", exchange);
+ commit.awaitRun();
+ own.awaitRun();
+
+ // the commit of the parent stays in the shared unit of work, and runs
when the parent is done
+ assertEquals(List.of(
+ "unitOfWork e1 u1 [commit]",
+ "added u1 commit",
+ "run e1 commit complete"),
+ recorder.lines(commit));
+ // the part hands over the on completion it added, which runs with the
copy
+ assertEquals(List.of(
+ "added u1 own",
+ "handover e1 u1 -> e2 [own]",
+ "unitOfWork e2 u2 [own]",
+ "added u2 own",
+ "run e2 own complete"),
+ recorder.lines(own));
+ // the part was given the unit of work of the parent
+ Event part = recorder.events.stream()
+ .filter(e -> e.type == Type.UNIT_OF_WORK && e.exchange !=
exchange
+ && e.unitOfWork == recorder.events.peek().unitOfWork)
+ .findFirst().orElseThrow();
+ assertTrue(part.list.isEmpty());
+ assertEquals(1, commit.runs.get());
+ assertEquals(1, own.runs.get());
+ }
+
+ @Test
+ void testVetoedHandoverIsNotReported() throws Exception {
+ context.getCamelContextExtension().addOnCompletionListener(recorder);
+ Commit commit = new VetoedCommit("vetoed");
+
+ Exchange exchange = createExchangeWithBody("A");
+ exchange.getExchangeExtension().addOnCompletion(commit);
+ template.send("direct:start", exchange);
+ commit.awaitRun();
+
+ // it refuses the handover to the copy sent to seda, so it stays and
runs with the exchange
+ assertEquals(List.of(
+ "unitOfWork e1 u1 [vetoed]",
+ "added u1 vetoed",
+ "run e1 vetoed complete"),
+ recorder.lines(commit));
+ assertEquals(1, commit.runs.get());
+ // nothing moved, so there is no handover event (and never one with an
empty list)
+ assertTrue(recorder.events.stream().noneMatch(e -> e.type ==
Type.HANDOVER && e.exchange == exchange));
+ assertTrue(recorder.events.stream().noneMatch(e -> e.list != null &&
e.list.isEmpty() && e.type != Type.UNIT_OF_WORK));
+ }
+
+ @Test
+ void testRemoved() {
+ context.getCamelContextExtension().addOnCompletionListener(recorder);
+ Commit removed = new Commit("removed");
+
+ Exchange exchange = createExchangeWithBody("A");
+ exchange.setProperty("removed", removed);
+ template.send("direct:remove", exchange);
+
+ assertEquals(List.of(
+ "added u1 removed",
+ "removed u1 removed"),
+ recorder.lines(removed));
+ assertEquals(0, removed.runs.get());
+ }
+
+ @Test
+ void testTakenAndRunBySomeoneElse() {
+ context.getCamelContextExtension().addOnCompletionListener(recorder);
+ Commit commit = new Commit("commit");
+
+ Exchange exchange = createExchangeWithBody("A");
+ exchange.getExchangeExtension().addOnCompletion(commit);
+ // as the aggregator, the enrichers and onCompletion do with an
exchange that has no unit of work
+ List<Synchronization> taken =
exchange.getExchangeExtension().handoverCompletions();
+ exchange.setException(new IllegalStateException("Forced"));
+ UnitOfWorkHelper.doneSynchronizations(exchange, taken);
+
+ assertEquals(List.of(
+ "taken e1 [commit]",
+ "run e1 commit failure"),
+ recorder.lines(commit));
+ assertEquals(1, commit.runs.get());
+ // the listener gets a view of the list, which it cannot change
+ Event event = recorder.events.peek();
+ assertThrows(UnsupportedOperationException.class, () ->
event.list.clear());
+ assertEquals(1, taken.size());
+ }
+
+ @Test
+ void testNoListener() throws Exception {
+
assertTrue(context.getCamelContextExtension().getOnCompletionListeners().isEmpty());
+ Commit commit = new Commit("commit");
+
+ Exchange exchange = createExchangeWithBody("A");
+ exchange.getExchangeExtension().addOnCompletion(commit);
+ template.send("direct:start", exchange);
+ commit.awaitRun();
+
+ assertEquals(1, commit.runs.get());
+ }
+
+ @Test
+ void testRemovedListenerIsNotCalled() throws Exception {
+ context.getCamelContextExtension().addOnCompletionListener(recorder);
+ // an equal listener is added once
+ context.getCamelContextExtension().addOnCompletionListener(recorder);
+ assertEquals(List.of(recorder),
context.getCamelContextExtension().getOnCompletionListeners());
+
context.getCamelContextExtension().removeOnCompletionListener(recorder);
+
assertTrue(context.getCamelContextExtension().getOnCompletionListeners().isEmpty());
+
+ Commit commit = new Commit("commit");
+ Exchange exchange = createExchangeWithBody("A");
+ exchange.getExchangeExtension().addOnCompletion(commit);
+ template.send("direct:start", exchange);
+ commit.awaitRun();
+
+ assertTrue(recorder.events.isEmpty());
+ assertEquals(1, commit.runs.get());
+ }
+
+ @Test
+ void testFailingListenerDoesNotChangeTheRouting() throws Exception {
+ OnCompletionListener failing = new OnCompletionListener() {
+ @Override
+ public void onUnitOfWork(Exchange exchange, UnitOfWork unitOfWork,
List<Synchronization> moved) {
+ throw new IllegalStateException("Forced");
+ }
+
+ @Override
+ public void onCompletionAdded(UnitOfWork unitOfWork,
Synchronization onCompletion) {
+ throw new IllegalStateException("Forced");
+ }
+
+ @Override
+ public void onCompletionHandover(
+ Exchange from, UnitOfWork fromUnitOfWork, Exchange to,
List<Synchronization> moved) {
+ throw new IllegalStateException("Forced");
+ }
+
+ @Override
+ public void onCompletionRemoved(UnitOfWork unitOfWork,
Synchronization onCompletion) {
+ throw new IllegalStateException("Forced");
+ }
+
+ @Override
+ public void onCompletionTaken(Exchange from, List<Synchronization>
taken) {
+ throw new IllegalStateException("Forced");
+ }
+
+ @Override
+ public void onCompletionRun(Exchange exchange, Synchronization
onCompletion, boolean failure) {
+ throw new IllegalStateException("Forced");
+ }
+ };
+ context.getCamelContextExtension().addOnCompletionListener(failing);
+ context.getCamelContextExtension().addOnCompletionListener(recorder);
+ Commit commit = new Commit("commit");
+
+ Exchange exchange = createExchangeWithBody("A");
+ exchange.getExchangeExtension().addOnCompletion(commit);
+ template.send("direct:start", exchange);
+ commit.awaitRun();
+
+ assertFalse(exchange.isFailed());
+ assertEquals(1, commit.runs.get());
+ assertEquals(6, recorder.lines(commit).size());
+
+ // also when it fails on a removal and on a take
+ Commit removed = new Commit("removed");
+ Exchange other = createExchangeWithBody("B");
+ other.setProperty("removed", removed);
+ template.send("direct:remove", other);
+ assertFalse(other.isFailed());
+ assertEquals(List.of("added u1 removed", "removed u1 removed"),
recorder.lines(removed));
+
+ Commit taken = new Commit("taken");
+ Exchange third = createExchangeWithBody("C");
+ third.getExchangeExtension().addOnCompletion(taken);
+ assertEquals(List.of(taken),
third.getExchangeExtension().handoverCompletions());
+ assertEquals(List.of("taken e1 [taken]"), recorder.lines(taken));
+ }
+
+ @Test
+ void testUnitOfWorkThatDoesNotTellWhatMoved() {
+ List<String> calls = new ArrayList<>();
+ Exchange target = createExchangeWithBody("B");
+ InvocationHandler handler = (proxy, method, args) -> {
+ if (method.isDefault()) {
+ return InvocationHandler.invokeDefault(proxy, method, args);
+ }
+ calls.add(method.getName() + " " + args.length);
+ return null;
+ };
+ UnitOfWork custom = (UnitOfWork)
Proxy.newProxyInstance(getClass().getClassLoader(),
+ new Class<?>[] { UnitOfWork.class }, handler);
+
+ // a custom unit of work that does not implement the new method hands
over as before, and tells nothing
+ assertNull(custom.handoverSynchronizationWithResult(target, null));
+ assertNull(custom.handoverSynchronizationWithResult(target, s ->
true));
+ assertEquals(List.of("handoverSynchronization 1",
"handoverSynchronization 2"), calls);
+ }
+
+ enum Type {
+ UNIT_OF_WORK,
+ ADDED,
+ REMOVED,
+ HANDOVER,
+ TAKEN,
+ RUN
+ }
+
+ record Event(Type type, Exchange exchange, UnitOfWork unitOfWork, Exchange
to, List<Synchronization> list,
+ Synchronization onCompletion, boolean failure) {
+ }
+
+ /**
+ * An on completion that counts its runs.
+ */
+ static class Commit implements Synchronization {
+ private final String name;
+ final AtomicInteger runs = new AtomicInteger();
+ private final CountDownLatch run = new CountDownLatch(1);
+
+ Commit(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public void onComplete(Exchange exchange) {
+ runs.incrementAndGet();
+ run.countDown();
+ }
+
+ @Override
+ public void onFailure(Exchange exchange) {
+ runs.incrementAndGet();
+ run.countDown();
+ }
+
+ void awaitRun() throws InterruptedException {
+ assertTrue(run.await(WAIT_SECONDS, TimeUnit.SECONDS), name + " did
not run");
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+ }
+
+ /**
+ * An on completion that refuses to be handed over.
+ */
+ static final class VetoedCommit extends Commit implements
SynchronizationVetoable {
+ VetoedCommit(String name) {
+ super(name);
+ }
+
+ @Override
+ public boolean allowHandover() {
+ return false;
+ }
+
+ @Override
+ public void beforeHandover(Exchange target) {
+ // not handed over
+ }
+ }
+
+ /**
+ * Records the events, and renders those of one on completion as lines,
with the exchanges and the units of work
+ * named in the order they first appear in those lines.
+ */
+ static final class Recorder implements OnCompletionListener {
+ final Queue<Event> events = new ConcurrentLinkedQueue<>();
+
+ @Override
+ public void onUnitOfWork(Exchange exchange, UnitOfWork unitOfWork,
List<Synchronization> moved) {
+ events.add(new Event(Type.UNIT_OF_WORK, exchange, unitOfWork,
null, moved, null, false));
+ }
+
+ @Override
+ public void onCompletionAdded(UnitOfWork unitOfWork, Synchronization
onCompletion) {
+ events.add(new Event(Type.ADDED, null, unitOfWork, null, null,
onCompletion, false));
+ }
+
+ @Override
+ public void onCompletionRemoved(UnitOfWork unitOfWork, Synchronization
onCompletion) {
+ events.add(new Event(Type.REMOVED, null, unitOfWork, null, null,
onCompletion, false));
+ }
+
+ @Override
+ public void onCompletionHandover(
+ Exchange from, UnitOfWork fromUnitOfWork, Exchange to,
List<Synchronization> moved) {
+ events.add(new Event(Type.HANDOVER, from, fromUnitOfWork, to,
moved, null, false));
+ }
+
+ @Override
+ public void onCompletionTaken(Exchange from, List<Synchronization>
taken) {
+ events.add(new Event(Type.TAKEN, from, null, null, taken, null,
false));
+ }
+
+ @Override
+ public void onCompletionRun(Exchange exchange, Synchronization
onCompletion, boolean failure) {
+ events.add(new Event(Type.RUN, exchange, null, null, null,
onCompletion, failure));
+ }
+
+ List<String> lines(Synchronization onCompletion) {
+ Predicate<Event> about = e -> e.onCompletion == onCompletion
+ || e.list != null && e.list.stream().anyMatch(s -> s ==
onCompletion);
+ Map<Object, String> names = new IdentityHashMap<>();
+ List<String> lines = new ArrayList<>();
+ for (Event e : events) {
+ if (!about.test(e)) {
+ continue;
+ }
+ String line = switch (e.type) {
+ case UNIT_OF_WORK -> "unitOfWork " + name(names,
e.exchange, "e") + " " + name(names, e.unitOfWork, "u")
+ + " " + e.list;
+ case ADDED -> "added " + name(names, e.unitOfWork, "u") +
" " + e.onCompletion;
+ case REMOVED -> "removed " + name(names, e.unitOfWork,
"u") + " " + e.onCompletion;
+ case HANDOVER -> "handover " + name(names, e.exchange,
"e") + " " + name(names, e.unitOfWork, "u")
+ + " -> " + name(names, e.to, "e") + " " +
e.list;
+ case TAKEN -> "taken " + name(names, e.exchange, "e") + "
" + e.list;
+ case RUN -> "run " + name(names, e.exchange, "e") + " " +
e.onCompletion + " "
+ + (e.failure ? "failure" : "complete");
+ };
+ lines.add(line);
+ }
+ return lines;
+ }
+
+ private static String name(Map<Object, String> names, Object o, String
prefix) {
+ if (o == null) {
+ return "-";
+ }
+ return names.computeIfAbsent(o, k -> prefix +
(names.keySet().stream()
+ .filter(n -> names.get(n).startsWith(prefix)).count() +
1));
+ }
+ }
+}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java
b/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java
index d4a06e789946..9b4cd0d75a1f 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java
@@ -17,6 +17,7 @@
package org.apache.camel.support;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.EnumMap;
import java.util.HashMap;
import java.util.LinkedHashMap;
@@ -24,6 +25,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Predicate;
import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
@@ -38,6 +40,7 @@ import org.apache.camel.Message;
import org.apache.camel.MessageHistory;
import org.apache.camel.Route;
import org.apache.camel.SafeCopyProperty;
+import org.apache.camel.spi.OnCompletionListener;
import org.apache.camel.spi.Synchronization;
import org.apache.camel.spi.UnitOfWork;
import org.apache.camel.spi.VariableRepository;
@@ -867,25 +870,49 @@ abstract class AbstractExchange implements Exchange,
ExchangeExtension {
@Override
public void handoverCompletions(Exchange target) {
+ final List<OnCompletionListener> listeners =
OnCompletionListenerHelper.getListeners(context);
+ final boolean listen = !listeners.isEmpty();
if (onCompletions != null) {
+ final List<Synchronization> moved = listen ? new
ArrayList<>(onCompletions) : null;
for (Synchronization onCompletion : onCompletions) {
target.getExchangeExtension().addOnCompletion(onCompletion);
}
onCompletions.clear();
onCompletions = null;
+ if (listen && !moved.isEmpty()) {
+ OnCompletionListenerHelper.onCompletionHandover(listeners,
this, null, target, moved);
+ }
} else if (unitOfWork != null) {
- if (sharedUnitOfWork == null || unitOfWork !=
sharedUnitOfWork.unitOfWork) {
- unitOfWork.handoverSynchronization(target);
+ final UnitOfWork from = unitOfWork;
+ if (sharedUnitOfWork == null || from !=
sharedUnitOfWork.unitOfWork) {
+ if (listen) {
+ handedOver(listeners, from, target,
from.handoverSynchronizationWithResult(target, null));
+ } else {
+ from.handoverSynchronization(target);
+ }
} else if (sharedUnitOfWork.completions != null) {
// the unit of work is shared with the parent exchange
(shareUnitOfWork), so hand over only the on
// completions this exchange added, and not those of the
parent exchange (such as its consumer's commit)
final List<Synchronization> own = sharedUnitOfWork.completions;
sharedUnitOfWork.completions = null;
- unitOfWork.handoverSynchronization(target, s ->
containsInstance(own, s));
+ final Predicate<Synchronization> filter = s ->
containsInstance(own, s);
+ if (listen) {
+ handedOver(listeners, from, target,
from.handoverSynchronizationWithResult(target, filter));
+ } else {
+ from.handoverSynchronization(target, filter);
+ }
}
}
}
+ private void handedOver(
+ List<OnCompletionListener> listeners, UnitOfWork from, Exchange
target, List<Synchronization> moved) {
+ // moved is null when the unit of work does not tell what moved
+ if (moved == null || !moved.isEmpty()) {
+ OnCompletionListenerHelper.onCompletionHandover(listeners, this,
from, target, moved);
+ }
+ }
+
private static boolean containsInstance(List<Synchronization> list,
Synchronization synchronization) {
for (Synchronization s : list) {
if (s == synchronization) {
@@ -902,6 +929,10 @@ abstract class AbstractExchange implements Exchange,
ExchangeExtension {
answer = new ArrayList<>(onCompletions);
onCompletions.clear();
onCompletions = null;
+ final List<OnCompletionListener> listeners =
OnCompletionListenerHelper.getListeners(context);
+ if (!listeners.isEmpty() && !answer.isEmpty()) {
+ OnCompletionListenerHelper.onCompletionTaken(listeners, this,
answer);
+ }
}
return answer;
}
@@ -913,6 +944,14 @@ abstract class AbstractExchange implements Exchange,
ExchangeExtension {
// no longer uses the shared unit of work (such as a copy routed
on its own)
sharedUnitOfWork = null;
}
+ if (unitOfWork != null) {
+ final List<OnCompletionListener> listeners =
OnCompletionListenerHelper.getListeners(context);
+ if (!listeners.isEmpty()) {
+ // before the on completions move, so the adds of the move
come after it
+ OnCompletionListenerHelper.onUnitOfWork(listeners, this,
unitOfWork,
+ onCompletions != null ? new ArrayList<>(onCompletions)
: Collections.emptyList());
+ }
+ }
if (unitOfWork != null && onCompletions != null) {
for (Synchronization onCompletion : onCompletions) {
unitOfWork.addSynchronization(onCompletion);
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/OnCompletionListenerHelper.java
b/core/camel-support/src/main/java/org/apache/camel/support/OnCompletionListenerHelper.java
new file mode 100644
index 000000000000..389bb9741a1e
--- /dev/null
+++
b/core/camel-support/src/main/java/org/apache/camel/support/OnCompletionListenerHelper.java
@@ -0,0 +1,145 @@
+/*
+ * 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.support;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.WeakHashMap;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedCamelContext;
+import org.apache.camel.spi.OnCompletionListener;
+import org.apache.camel.spi.Synchronization;
+import org.apache.camel.spi.UnitOfWork;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Calls the {@link OnCompletionListener}s, for the places in Camel that
change or run the on completions of an
+ * exchange. A place gets the listeners with {@link
#getListeners(CamelContext)} and calls them only when the list is
+ * not empty, so with no listener installed it only checks that the list is
empty. What a listener throws is caught and
+ * logged at WARN once per listener (then at DEBUG), so it cannot change the
routing.
+ * <p/>
+ * <b>Important:</b> This API is NOT intended for Camel end users, but used
internally by Camel itself.
+ *
+ * @since 4.24
+ */
+public final class OnCompletionListenerHelper {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(OnCompletionListenerHelper.class);
+
+ // the listeners that already failed once, so they are logged at WARN once
(weak keys: a listener is not kept)
+ private static final Map<OnCompletionListener, Boolean> FAILED_LISTENERS
+ = Collections.synchronizedMap(new WeakHashMap<>());
+
+ private OnCompletionListenerHelper() {
+ }
+
+ /**
+ * The listeners installed on the context.
+ *
+ * @return the listeners (a view, not a copy), or an empty list if none is
installed or there is no context
+ */
+ public static List<OnCompletionListener> getListeners(CamelContext
context) {
+ final ExtendedCamelContext ecc = context != null ?
context.getCamelContextExtension() : null;
+ return ecc != null ? ecc.getOnCompletionListeners() :
Collections.emptyList();
+ }
+
+ public static void onUnitOfWork(
+ List<OnCompletionListener> listeners, Exchange exchange,
UnitOfWork unitOfWork, List<Synchronization> moved) {
+ final List<Synchronization> view = Collections.unmodifiableList(moved);
+ for (OnCompletionListener listener : listeners) {
+ try {
+ listener.onUnitOfWork(exchange, unitOfWork, view);
+ } catch (Throwable e) {
+ failed(listener, "onUnitOfWork", e);
+ }
+ }
+ }
+
+ public static void onCompletionAdded(
+ List<OnCompletionListener> listeners, UnitOfWork unitOfWork,
Synchronization onCompletion) {
+ for (OnCompletionListener listener : listeners) {
+ try {
+ listener.onCompletionAdded(unitOfWork, onCompletion);
+ } catch (Throwable e) {
+ failed(listener, "onCompletionAdded", e);
+ }
+ }
+ }
+
+ public static void onCompletionRemoved(
+ List<OnCompletionListener> listeners, UnitOfWork unitOfWork,
Synchronization onCompletion) {
+ for (OnCompletionListener listener : listeners) {
+ try {
+ listener.onCompletionRemoved(unitOfWork, onCompletion);
+ } catch (Throwable e) {
+ failed(listener, "onCompletionRemoved", e);
+ }
+ }
+ }
+
+ public static void onCompletionHandover(
+ List<OnCompletionListener> listeners, Exchange from, UnitOfWork
fromUnitOfWork, Exchange to,
+ List<Synchronization> moved) {
+ final List<Synchronization> view = moved != null ?
Collections.unmodifiableList(moved) : null;
+ for (OnCompletionListener listener : listeners) {
+ try {
+ listener.onCompletionHandover(from, fromUnitOfWork, to, view);
+ } catch (Throwable e) {
+ failed(listener, "onCompletionHandover", e);
+ }
+ }
+ }
+
+ public static void onCompletionTaken(List<OnCompletionListener> listeners,
Exchange from, List<Synchronization> taken) {
+ final List<Synchronization> view = Collections.unmodifiableList(taken);
+ for (OnCompletionListener listener : listeners) {
+ try {
+ listener.onCompletionTaken(from, view);
+ } catch (Throwable e) {
+ failed(listener, "onCompletionTaken", e);
+ }
+ }
+ }
+
+ public static void onCompletionRun(
+ List<OnCompletionListener> listeners, Exchange exchange,
Synchronization onCompletion, boolean failure) {
+ for (OnCompletionListener listener : listeners) {
+ try {
+ listener.onCompletionRun(exchange, onCompletion, failure);
+ } catch (Throwable e) {
+ failed(listener, "onCompletionRun", e);
+ }
+ }
+ }
+
+ private static void failed(OnCompletionListener listener, String method,
Throwable cause) {
+ try {
+ if (FAILED_LISTENERS.put(listener, Boolean.TRUE) == null) {
+ LOG.warn("OnCompletionListener {} failed in {} (logged once
for this listener, then at DEBUG)", listener,
+ method, cause);
+ } else {
+ LOG.debug("OnCompletionListener {} failed in {}", listener,
method, cause);
+ }
+ } catch (Throwable t) {
+ // a listener cannot change the routing, also when logging it
fails (such as its toString)
+ }
+ }
+}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/UnitOfWorkHelper.java
b/core/camel-support/src/main/java/org/apache/camel/support/UnitOfWorkHelper.java
index 5ef62a1e143a..8fb1f769ffbc 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/UnitOfWorkHelper.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/UnitOfWorkHelper.java
@@ -22,6 +22,7 @@ import java.util.List;
import org.apache.camel.Exchange;
import org.apache.camel.Route;
+import org.apache.camel.spi.OnCompletionListener;
import org.apache.camel.spi.Synchronization;
import org.apache.camel.spi.SynchronizationRouteAware;
import org.apache.camel.spi.UnitOfWork;
@@ -76,6 +77,7 @@ public final class UnitOfWorkHelper {
return;
}
+ final List<OnCompletionListener> listeners =
OnCompletionListenerHelper.getListeners(exchange.getContext());
if (synchronizations.size() > 1) {
// work on a copy of the list (unless it is private) to avoid any
modification which may cause
// ConcurrentModificationException
@@ -85,15 +87,19 @@ public final class UnitOfWorkHelper {
// invoke synchronization callbacks
for (Synchronization synchronization : ordered) {
- doneSynchronization(synchronization, exchange, failed);
+ doneSynchronization(synchronization, exchange, failed,
listeners);
}
} else {
// there are only 1 synchronization to done
- doneSynchronization(synchronizations.get(0), exchange,
exchange.isFailed());
+ doneSynchronization(synchronizations.get(0), exchange,
exchange.isFailed(), listeners);
}
}
- private static void doneSynchronization(Synchronization synchronization,
Exchange exchange, boolean failed) {
+ private static void doneSynchronization(
+ Synchronization synchronization, Exchange exchange, boolean
failed, List<OnCompletionListener> listeners) {
+ if (!listeners.isEmpty()) {
+ OnCompletionListenerHelper.onCompletionRun(listeners, exchange,
synchronization, failed);
+ }
try {
if (failed) {
LOG.trace("Invoking synchronization.onFailure: {} with {}",
synchronization, exchange);
diff --git a/design/exchange-sanitizer.adoc b/design/exchange-sanitizer.adoc
index edb23d273383..6587422108dc 100644
--- a/design/exchange-sanitizer.adoc
+++ b/design/exchange-sanitizer.adoc
@@ -387,9 +387,13 @@ public interface OnCompletionListener {
// an on completion is added to a unit of work
default void onCompletionAdded(UnitOfWork unitOfWork, Synchronization
onCompletion) { }
- // on completions are handed over from one exchange (and its unit of work,
if any) to another exchange
+ // an on completion is removed from a unit of work, so it does not run
+ default void onCompletionRemoved(UnitOfWork unitOfWork, Synchronization
onCompletion) { }
+
+ // on completions are handed over from one exchange (and its unit of work,
if any) to another exchange; moved is
+ // null when a custom unit of work does not tell what moved
default void onCompletionHandover(
- Exchange from, @Nullable UnitOfWork fromUnitOfWork, Exchange to,
List<Synchronization> moved) { }
+ Exchange from, @Nullable UnitOfWork fromUnitOfWork, Exchange to,
@Nullable List<Synchronization> moved) { }
// on completions are taken out of an exchange as a list, to be run by
someone else
default void onCompletionTaken(Exchange from, List<Synchronization> taken)
{ }
@@ -418,9 +422,9 @@ They are installed through `ExtendedCamelContext`, as the
`StepHook` is, with de
of the order of work); the on completion listener comes with R2 (step 3).
The API changes in short: the two interfaces; the six methods on
`ExtendedCamelContext`; and, for R2, a new default
-method on `UnitOfWork` that hands over and returns what moved
(`handoverSynchronization(Exchange from, Exchange
-target, Predicate<Synchronization> filter)`), as today only
`DefaultUnitOfWork` knows which on completions a
-`SynchronizationVetoable` kept, and only the exchange knows that it is the one
handing over.
+method on `UnitOfWork` that hands over and returns what moved
(`handoverSynchronizationWithResult(Exchange target,
+Predicate<Synchronization> filter)`), as today only `DefaultUnitOfWork` knows
which on completions a
+`SynchronizationVetoable` kept.
Rules for listeners, in the Javadoc:
@@ -488,12 +492,14 @@ wrapped callback; that is a development cost, measured
before the first pull req
*On completions.* The on completion listener is called where Camel changes or
runs on completions today:
-* `onUnitOfWork` in `setUnitOfWork` of the exchange (in `AbstractExchange` and
`ExtendedExchangeExtension`), with the
- on completions it moves into the unit of work;
+* `onUnitOfWork` in `setUnitOfWork` of the exchange (in `AbstractExchange`;
`ExtendedExchangeExtension` is no longer
+ used since CAMEL-23686), with the on completions it moves into the unit of
work;
* `onCompletionAdded` in `DefaultUnitOfWork.addSynchronization`. An add that
is part of a move or of a handover is
reported there too. The listener tells them apart: `onUnitOfWork` is fired
before the move, and `onCompletionHandover`
right after the handover, on the same thread, so the adds in between belong
to them. Adds to an exchange's own list
(before it has a unit of work) are not reported; they show up in the move;
+* `onCompletionRemoved` in `DefaultUnitOfWork.removeSynchronization`, when it
removed one, as a removed commit does not
+ run (see `Synchronization`);
* `onCompletionHandover` in `handoverCompletions(target)` of the exchange,
which knows the exchange that hands over and,
through the new `UnitOfWork` method, what moved;
* `onCompletionTaken` in `handoverCompletions()` without a target;
