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 931bee0a80d1 CAMEL-26331: camel-core - one hook for every send to an
endpoint, that can stop, delay or change the send (#28598)
931bee0a80d1 is described below
commit 931bee0a80d1aec7e47d6575e2aec7a3a17416c8
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Oct 28 17:18:34 2026 +0530
CAMEL-26331: camel-core - one hook for every send to an endpoint, that can
stop, delay or change the send (#28598)
The six places that fire ExchangeSendingEvent (SendProcessor,
DefaultProducerCache send and
doInAsyncProducer, EventNotifierCallback, MulticastProcessor for the
recipient list,
EventNotifierProducer) repeated the same code: fire the sending event,
start a StopWatch, call the
producer, fire the sent event. They now call one helper, SendHelper in
camel-support.
The helper runs a new general SPI, org.apache.camel.spi.SendHook, installed
with
ExtendedCamelContext#setSendHook, around the call of the producer. Before
the producer it can
change the message, stop the send, skip it or delay it; after the producer
it can change the
reply. It runs inside the error handler where the producer is called (per
recipient for the
recipient list, per step for the routing slip), so a stopped send is a
failure for onException,
redelivery and the dead letter channel. A delay waits on the thread that
sends via
TimeSource.sleep, so a virtual clock can drive it. Other negative values
are reserved. The
callback is called exactly once on every path.
Without a hook the cost is one volatile read per send, and the events,
timing and error handling
are unchanged. First code step of camel-chaos (CAMEL-26330,
design/chaos.adoc).
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/ExtendedCamelContext.java | 25 +
.../java/org/apache/camel/spi/ProducerCache.java | 3 +
.../main/java/org/apache/camel/spi/SendHook.java | 114 +++++
.../impl/engine/DefaultCamelContextExtension.java | 12 +
.../camel/impl/ExtendedCamelContextConfigurer.java | 6 +
.../camel/processor/EventNotifierProducer.java | 28 +-
.../apache/camel/processor/MulticastProcessor.java | 54 +-
.../org/apache/camel/processor/SendProcessor.java | 21 +-
.../apache/camel/processor/SendHookStressTest.java | 173 +++++++
.../org/apache/camel/processor/SendHookTest.java | 500 +++++++++++++++++++
.../org/apache/camel/support/SendHelperTest.java | 544 +++++++++++++++++++++
.../java/org/apache/camel/support/SendHelper.java | 348 +++++++++++++
.../camel/support/cache/DefaultProducerCache.java | 38 +-
.../camel/support/cache/EventNotifierCallback.java | 14 +-
14 files changed, 1786 insertions(+), 94 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 55e411d59094..e2eede3274a2 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
@@ -42,6 +42,7 @@ import org.apache.camel.spi.ReactiveExecutor;
import org.apache.camel.spi.Registry;
import org.apache.camel.spi.RouteController;
import org.apache.camel.spi.RouteStartupOrder;
+import org.apache.camel.spi.SendHook;
import org.apache.camel.spi.StartupStepRecorder;
import org.jspecify.annotations.Nullable;
@@ -581,6 +582,30 @@ public interface ExtendedCamelContext {
*/
void setEventNotificationApplicable(boolean eventNotificationApplicable);
+ /**
+ * Gets the {@link SendHook} that runs around the sends to the producer of
an endpoint that fire the
+ * {@link org.apache.camel.spi.CamelEvent.ExchangeSendingEvent}.
+ *
+ * @return the hook, or <tt>null</tt> if no hook is installed
+ * @since 4.24
+ */
+ default @Nullable SendHook getSendHook() {
+ return null;
+ }
+
+ /**
+ * Installs a {@link SendHook} that runs around the sends to the producer
of an endpoint that fire the
+ * {@link org.apache.camel.spi.CamelEvent.ExchangeSendingEvent}, or
removes it with <tt>null</tt>. There is one
+ * hook; a new hook replaces the one installed before. Sends that are
already in the hook complete with the hook
+ * that started them.
+ *
+ * @param sendHook the hook, or <tt>null</tt> to remove it
+ * @since 4.24
+ */
+ default void setSendHook(@Nullable SendHook sendHook) {
+ throw new UnsupportedOperationException("This CamelContext does not
support a SendHook");
+ }
+
/**
* Internal {@link RouteController} that are only used internally by Camel
to perform basic route operations. Do not
* use this as end user.
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/ProducerCache.java
b/core/camel-api/src/main/java/org/apache/camel/spi/ProducerCache.java
index ac338738b22f..74971d0e2cb5 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/ProducerCache.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/ProducerCache.java
@@ -156,6 +156,9 @@ public interface ProducerCache extends Service {
* <p/>
* Using this callback as a template pattern ensures that Camel handles
the resource handling and will start and
* stop the given producer, to avoid resource leaks.
+ * <p/>
+ * When a {@link SendHook} is installed, the producer given to the
callback is a producer that calls the producer of
+ * the endpoint with the hook around it.
*/
interface AsyncProducerCallback {
diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/SendHook.java
b/core/camel-api/src/main/java/org/apache/camel/spi/SendHook.java
new file mode 100644
index 000000000000..8db73d822927
--- /dev/null
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/SendHook.java
@@ -0,0 +1,114 @@
+/*
+ * 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.Endpoint;
+import org.apache.camel.Exchange;
+
+/**
+ * A hook that runs around the sends to the producer of an {@link Endpoint}
that fire the
+ * {@link CamelEvent.ExchangeSendingEvent}: the sends of a route ({@code to},
{@code toD}, {@code wireTap},
+ * {@code enrich}, {@code recipientList}, {@code routingSlip}, {@code
dynamicRouter}), of a
+ * {@link org.apache.camel.ProducerTemplate} and of a deferred producer. Code
that gets a producer itself and calls it
+ * (for example with {@link ProducerCache#acquireProducer(Endpoint)}) fires no
such event and does not run the hook.
+ * Before the producer is called, the hook can change the message, stop the
send, skip it or delay it. After the
+ * producer is done, it can change the reply.
+ * <p/>
+ * The hook is installed with {@link
org.apache.camel.ExtendedCamelContext#setSendHook(SendHook)}. There is one hook
per
+ * {@link org.apache.camel.CamelContext}; to use more than one, combine them
in one hook. Without a hook, a send costs
+ * one check whether a hook is installed.
+ * <p/>
+ * <b>Where it runs.</b> The hook runs right around the call of the producer,
inside the error handler that handles a
+ * failure of that producer: the error handler of the route for {@code to},
{@code toD} and {@code enrich}, the error
+ * handler of the tapped copy for {@code wireTap}, and the error handler of
the recipient or of the step for
+ * {@code recipientList}, {@code routingSlip} and {@code dynamicRouter}. So an
exception set by
+ * {@link #beforeSend(Exchange, Endpoint)} is handled as a failure of the
producer: {@code onException}, redelivery and
+ * the dead letter channel apply to it, and each redelivery runs the hook
again; inside {@code doTry}, {@code doCatch}
+ * gets it. It runs after the {@link CamelEvent.ExchangeSendingEvent} and
before the
+ * {@link CamelEvent.ExchangeSentEvent}, so the time taken of the send
includes a delay of the hook. A send from a
+ * {@link org.apache.camel.ProducerTemplate} has no error handler: the
exception is returned to the caller.
+ * <p/>
+ * <b>Threads.</b> The hook is called by many threads at the same time, and
must be thread-safe.
+ * {@link #beforeSend(Exchange, Endpoint)} runs on the thread that sends,
which also waits for a delay.
+ * {@link #afterSend(Exchange, Endpoint)} runs on the thread that completes
the send, which can be another thread for an
+ * asynchronous producer.
+ * <p/>
+ * <b>Contract.</b> {@link #afterSend(Exchange, Endpoint)} is called exactly
once for every call of
+ * {@link #beforeSend(Exchange, Endpoint)} that returned: after the producer
is done, or right away when the producer is
+ * not called (the hook stopped or skipped the send, returned a reserved
value, or the wait for its delay was
+ * interrupted). It is also called when the producer throws an exception
instead of completing; the exception is then
+ * set on the exchange. When {@link #beforeSend(Exchange, Endpoint)} throws an
exception, the exception is set on the
+ * exchange, the producer is not called and {@link #afterSend(Exchange,
Endpoint)} is not called. When
+ * {@link #afterSend(Exchange, Endpoint)} throws an exception, it is set on
the exchange (or added as a suppressed
+ * exception when the exchange has failed already). In every case the send
completes exactly once.
+ *
+ * @since 4.24
+ */
+public interface SendHook {
+
+ /**
+ * Returned by {@link #beforeSend(Exchange, Endpoint)} to call the
producer right away.
+ */
+ long PROCEED = 0;
+
+ /**
+ * Returned by {@link #beforeSend(Exchange, Endpoint)} to complete the
send without calling the producer, as if the
+ * producer had succeeded. The exchange is left as the hook left it.
{@link #afterSend(Exchange, Endpoint)} is
+ * called, and the {@link CamelEvent.ExchangeSendingEvent} and {@link
CamelEvent.ExchangeSentEvent} are fired as for
+ * any other send.
+ */
+ long SKIP = -1;
+
+ /**
+ * Called before the producer of the endpoint is called.
+ * <p/>
+ * The hook can change the message of the exchange, which the producer
then receives. To stop the send it sets an
+ * exception on the exchange: the producer is not called, and the send
fails with that exception. (An exchange that
+ * already has an exception when it is sent, which is rare, is sent as
usual unless the hook sets another one.)
+ * <p/>
+ * A positive return value delays the call of the producer by that many
milliseconds: the thread that sends waits,
+ * as for a slow endpoint, and then calls the producer on the same thread.
That is whatever thread sends, also a
+ * thread of an asynchronous component on which the route goes on after
its producer. It waits with
+ * {@link org.apache.camel.util.TimeSource#sleep(long)}, so a test harness
that installs a virtual clock there runs
+ * the delay in virtual time. When the wait is interrupted, the producer
is not called and the send fails with the
+ * {@link InterruptedException}; the thread keeps its interrupted status.
As for any {@link InterruptedException} on
+ * an exchange, Camel marks the exchange as interrupted and stops routing
it, instead of handling the failure with
+ * {@code onException}, redelivery or the dead letter channel (as when the
wait of the redelivery error handler is
+ * interrupted).
+ *
+ * @param exchange the exchange that is sent
+ * @param endpoint the endpoint it is sent to
+ * @return {@link #PROCEED} to call the producer now, a delay in
milliseconds to wait before calling it, or
+ * {@link #SKIP} to complete the send without calling it.
Other negative values are reserved: the
+ * send fails with an {@link IllegalArgumentException},
without calling the producer (and
+ * {@link #afterSend(Exchange, Endpoint)} is called). The
value is ignored when the hook sets an
+ * exception on the exchange.
+ */
+ long beforeSend(Exchange exchange, Endpoint endpoint);
+
+ /**
+ * Called after the producer of the endpoint is done, or right away when
the producer is not called (see
+ * {@link #beforeSend(Exchange, Endpoint)}).
+ * <p/>
+ * The hook can change the reply (the message of the exchange), or set or
clear the exception of the exchange.
+ *
+ * @param exchange the exchange that was sent
+ * @param endpoint the endpoint it was sent to
+ */
+ void afterSend(Exchange exchange, Endpoint endpoint);
+
+}
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 b579e134dc19..9671e90ee62c 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
@@ -74,6 +74,7 @@ import org.apache.camel.spi.RestConfiguration;
import org.apache.camel.spi.RestRegistryFactory;
import org.apache.camel.spi.RouteController;
import org.apache.camel.spi.RouteStartupOrder;
+import org.apache.camel.spi.SendHook;
import org.apache.camel.spi.ShutdownStrategy;
import org.apache.camel.spi.StartupStepRecorder;
import org.apache.camel.spi.StreamCachingStrategy;
@@ -120,6 +121,7 @@ class DefaultCamelContextExtension implements
ExtendedCamelContext {
private volatile ManagementMBeanAssembler managementMBeanAssembler;
private volatile HeadersMapFactory headersMapFactory;
private volatile boolean eventNotificationApplicable;
+ private volatile SendHook sendHook;
private volatile CamelContextNameStrategy nameStrategy;
private volatile ManagementNameStrategy managementNameStrategy;
private volatile PropertiesComponent propertiesComponent;
@@ -553,6 +555,16 @@ class DefaultCamelContextExtension implements
ExtendedCamelContext {
this.eventNotificationApplicable = eventNotificationApplicable;
}
+ @Override
+ public SendHook getSendHook() {
+ return sendHook;
+ }
+
+ @Override
+ public void setSendHook(SendHook sendHook) {
+ this.sendHook = sendHook;
+ }
+
@Override
public FactoryFinder getDefaultFactoryFinder() {
return getFactoryFinder(FactoryFinder.DEFAULT_PATH);
diff --git
a/core/camel-core-engine/src/generated/java/org/apache/camel/impl/ExtendedCamelContextConfigurer.java
b/core/camel-core-engine/src/generated/java/org/apache/camel/impl/ExtendedCamelContextConfigurer.java
index 433ff9038a80..73d2e3bcb9f1 100644
---
a/core/camel-core-engine/src/generated/java/org/apache/camel/impl/ExtendedCamelContextConfigurer.java
+++
b/core/camel-core-engine/src/generated/java/org/apache/camel/impl/ExtendedCamelContextConfigurer.java
@@ -53,6 +53,8 @@ public class ExtendedCamelContextConfigurer extends
org.apache.camel.support.com
case "reactiveexecutor":
case "reactiveExecutor":
target.setReactiveExecutor(property(camelContext,
org.apache.camel.spi.ReactiveExecutor.class, value)); return true;
case "registry": target.setRegistry(property(camelContext,
org.apache.camel.spi.Registry.class, value)); return true;
+ case "sendhook":
+ case "sendHook": target.setSendHook(property(camelContext,
org.apache.camel.spi.SendHook.class, value)); return true;
case "startupsteprecorder":
case "startupStepRecorder":
target.setStartupStepRecorder(property(camelContext,
org.apache.camel.spi.StartupStepRecorder.class, value)); return true;
default: return false;
@@ -92,6 +94,8 @@ public class ExtendedCamelContextConfigurer extends
org.apache.camel.support.com
case "reactiveexecutor":
case "reactiveExecutor": return
org.apache.camel.spi.ReactiveExecutor.class;
case "registry": return org.apache.camel.spi.Registry.class;
+ case "sendhook":
+ case "sendHook": return org.apache.camel.spi.SendHook.class;
case "startupsteprecorder":
case "startupStepRecorder": return
org.apache.camel.spi.StartupStepRecorder.class;
default: return null;
@@ -132,6 +136,8 @@ public class ExtendedCamelContextConfigurer extends
org.apache.camel.support.com
case "reactiveexecutor":
case "reactiveExecutor": return target.getReactiveExecutor();
case "registry": return target.getRegistry();
+ case "sendhook":
+ case "sendHook": return target.getSendHook();
case "startupsteprecorder":
case "startupStepRecorder": return target.getStartupStepRecorder();
default: return null;
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/EventNotifierProducer.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/EventNotifierProducer.java
index 380f04e21661..bb2f32abc739 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/EventNotifierProducer.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/EventNotifierProducer.java
@@ -23,9 +23,8 @@ import org.apache.camel.Exchange;
import org.apache.camel.Producer;
import org.apache.camel.support.AsyncProcessorConverterHelper;
import org.apache.camel.support.DefaultAsyncProducer;
-import org.apache.camel.support.EventHelper;
+import org.apache.camel.support.SendHelper;
import org.apache.camel.support.service.ServiceHelper;
-import org.apache.camel.util.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -56,31 +55,10 @@ public final class EventNotifierProducer extends
DefaultAsyncProducer {
return true;
}
- final boolean sending =
EventHelper.notifyExchangeSending(exchange.getContext(), exchange,
getEndpoint());
- // record timing for sending the exchange using the producer
- StopWatch watch;
- if (sending) {
- watch = new StopWatch();
- } else {
- watch = null;
- }
-
try {
LOG.debug(">>>> {} {}", getEndpoint(), exchange);
- return producer.process(exchange, new AsyncCallback() {
- @Override
- public void done(boolean doneSync) {
- try {
- // emit event that the exchange was sent to the
endpoint
- if (watch != null) {
- long timeTaken = watch.taken();
-
EventHelper.notifyExchangeSent(exchange.getContext(), exchange, getEndpoint(),
timeTaken);
- }
- } finally {
- callback.done(doneSync);
- }
- }
- });
+ // emit the sending and sent events around the send (and run the
send hook if any)
+ return SendHelper.send(exchange, getEndpoint(), producer,
callback);
} catch (Exception throwable) {
exchange.setException(throwable);
callback.done(true);
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
index c96b75160450..9b54724c468b 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java
@@ -39,10 +39,10 @@ import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.AggregationStrategy;
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
+import org.apache.camel.AsyncProducer;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
import org.apache.camel.CamelExchangeException;
-import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.Navigate;
@@ -68,10 +68,10 @@ import
org.apache.camel.support.AsyncProcessorConverterHelper;
import org.apache.camel.support.AsyncProcessorSupport;
import org.apache.camel.support.CamelContextHelper;
import org.apache.camel.support.DefaultExchange;
-import org.apache.camel.support.EventHelper;
import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.LRUCacheFactory;
import org.apache.camel.support.PluginHelper;
+import org.apache.camel.support.SendHelper;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.CastUtils;
import org.apache.camel.util.IOHelper;
@@ -190,7 +190,7 @@ public class MulticastProcessor extends BaseProcessorSupport
private boolean shutdownAggregateExecutorService;
private final long timeout;
private final int cacheSize;
- private final Map<Processor, Processor> errorHandlers;
+ private final Map<Object, Processor> errorHandlers;
private final boolean shareUnitOfWork;
public MulticastProcessor(CamelContext camelContext, Route route,
Collection<Processor> processors) {
@@ -896,12 +896,7 @@ public class MulticastProcessor extends
BaseProcessorSupport
final Exchange e = pair.getExchange();
final Producer p = pair.getProducer();
if (p != null) {
- boolean sending =
EventHelper.notifyExchangeSending(e.getContext(), e, p.getEndpoint());
- if (sending) {
- watch = new StopWatch();
- } else {
- watch = null;
- }
+ watch = SendHelper.notifySending(e, p.getEndpoint());
} else {
watch = null;
}
@@ -919,12 +914,9 @@ public class MulticastProcessor extends
BaseProcessorSupport
// okay we are done, so notify the exchange was sent
final Producer producer = pair.getProducer();
- if (producer != null && watch != null) {
- long timeTaken = watch.taken();
- final Exchange e = pair.getExchange();
- Endpoint endpoint = producer.getEndpoint();
+ if (producer != null) {
// emit event that the exchange was sent to the endpoint
- EventHelper.notifyExchangeSent(e.getContext(), e, endpoint,
timeTaken);
+ SendHelper.notifySent(pair.getExchange(), producer.getEndpoint(),
watch);
}
}
@@ -1197,8 +1189,15 @@ public class MulticastProcessor extends
BaseProcessorSupport
// this is needed to support redelivery on that output alone and
not doing redelivery
// for the entire multicast block again which will start from
scratch again
+ // with a send hook installed, the output of the error handler of
a producer (a recipient of the recipient
+ // list) is the producer with the hook around it, and that error
handler is cached apart, so without a hook
+ // the error handler and its output are the same as when there is
no send hook at all
+ final boolean hookInstalled = key instanceof AsyncProducer
+ && camelContext.getCamelContextExtension().getSendHook()
!= null;
+ final Object cacheKey = hookInstalled ? new SendHookKey(key) : key;
+
// lookup cached first to reuse and preserve memory
- answer = errorHandlers != null ? errorHandlers.get(key) : null;
+ answer = errorHandlers != null ? errorHandlers.get(cacheKey) :
null;
if (answer != null) {
LOG.trace("Using existing error handler for: {}", key);
return answer;
@@ -1206,7 +1205,9 @@ public class MulticastProcessor extends
BaseProcessorSupport
LOG.trace("Creating error handler for: {}", key);
try {
- processor = wrapInErrorHandler(route, key);
+ // the producer with the hook around it looks up the hook on
every send, so it can be cached
+ Processor output = hookInstalled ?
SendHelper.newSendHookProducer((AsyncProducer) key) : key;
+ processor = wrapInErrorHandler(route, output);
// and wrap in unit of work processor so the copy exchange
also can run under UoW
answer = createUnitOfWorkProcessor(route, processor, exchange);
@@ -1215,7 +1216,7 @@ public class MulticastProcessor extends
BaseProcessorSupport
ServiceHelper.startService(answer);
if (errorHandlers != null) {
- errorHandlers.putIfAbsent(key, answer);
+ errorHandlers.putIfAbsent(cacheKey, answer);
}
} catch (Exception e) {
@@ -1223,12 +1224,29 @@ public class MulticastProcessor extends
BaseProcessorSupport
}
} else {
// and wrap in unit of work processor so the copy exchange also
can run under UoW
- answer = createUnitOfWorkProcessor(route, processor, exchange);
+ answer = createUnitOfWorkProcessor(route, withSendHook(processor),
exchange);
}
return answer;
}
+ /**
+ * Puts the send hook, if one is installed, around an output that is a
producer (a recipient of the recipient list),
+ * when the output is used for this exchange only.
+ */
+ private Processor withSendHook(Processor output) {
+ if (output instanceof AsyncProducer producer) {
+ return SendHelper.withSendHook(camelContext, producer);
+ }
+ return output;
+ }
+
+ /**
+ * The key of the cached error handler of a producer whose output has the
send hook around it.
+ */
+ private record SendHookKey(Processor producer) {
+ }
+
private Processor wrapInErrorHandler(Route route, Processor processor)
throws Exception {
// use the error handler from multicast and clone it to use the new
processor as its output
if (errorHandler instanceof ErrorHandlerSupport errorHandlerSupport) {
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/SendProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/SendProcessor.java
index 48f135653234..e3de9c00bc8e 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/SendProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/SendProcessor.java
@@ -39,6 +39,7 @@ import org.apache.camel.support.EndpointHelper;
import org.apache.camel.support.EventHelper;
import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.PluginHelper;
+import org.apache.camel.support.SendHelper;
import org.apache.camel.support.cache.DefaultProducerCache;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.ObjectHelper;
@@ -221,15 +222,9 @@ public class SendProcessor extends BaseProcessorSupport
// set property which endpoint we send to
exchange.setProperty(ExchangePropertyKey.TO_ENDPOINT,
destination.getEndpointUri());
- final boolean sending =
camelContext.getCamelContextExtension().isEventNotificationApplicable()
- && EventHelper.notifyExchangeSending(exchange.getContext(),
target, destination);
- // record timing for sending the exchange using the producer
- StopWatch watch;
- if (sending) {
- watch = new StopWatch();
- } else {
- watch = null;
- }
+ // emit event that the exchange is being sent to the endpoint, and
record timing for sending the exchange
+ final StopWatch watch =
camelContext.getCamelContextExtension().isEventNotificationApplicable()
+ ? SendHelper.notifySending(target, destination) : null;
// optimize to only create a new callback if really needed, otherwise
we can use the provided callback as-is
AsyncCallback ac = callback;
@@ -247,10 +242,7 @@ public class SendProcessor extends BaseProcessorSupport
// restore previous MEP
target.setPattern(existingPattern);
// emit event that the exchange was sent to the endpoint
- if (watch != null) {
- long timeTaken = watch.taken();
- EventHelper.notifyExchangeSent(target.getContext(),
target, destination, timeTaken);
- }
+ SendHelper.notifySent(target, destination, watch);
} finally {
callback.done(doneSync);
}
@@ -264,7 +256,8 @@ public class SendProcessor extends BaseProcessorSupport
}
LOG.debug(">>>> {} {}", destination, exchange);
- boolean sync = producer.process(exchange, ac);
+ // call the producer (with the send hook around it if any)
+ boolean sync = SendHelper.process(exchange, destination, producer,
ac);
if (!sync) {
// Deprecated since 4.19.0
EventHelper.notifyExchangeAsyncProcessingStartedEvent(camelContext, exchange);
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SendHookStressTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SendHookStressTest.java
new file mode 100644
index 000000000000..6fc199518110
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SendHookStressTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.processor.async.MyAsyncComponent;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeCompletedEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeFailedEvent;
+import org.apache.camel.spi.SendHook;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.EventNotifierSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Many concurrent sends through {@code to}, {@code toD}, {@code
recipientList} (with an asynchronous producer) and
+ * {@code routingSlip}, with a {@link SendHook} that randomly stops, skips,
delays or changes each send, and a dead
+ * letter channel with redelivery: every exchange completes exactly once, and
the hook runs after once for every before.
+ */
+class SendHookStressTest extends ContextTestSupport {
+
+ private static final int SENDERS = 8;
+ private static final int PER_SENDER = 250;
+
+ private final AtomicLong before = new AtomicLong();
+ private final AtomicLong after = new AtomicLong();
+ private final AtomicLong stopped = new AtomicLong();
+ private final AtomicLong delayed = new AtomicLong();
+ // the exchanges sent by the test, and how often each completed (or failed)
+ private final Map<String, AtomicInteger> completions = new
ConcurrentHashMap<>();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) {
+ if (event instanceof ExchangeCompletedEvent || event
instanceof ExchangeFailedEvent) {
+ AtomicInteger count =
completions.get(((CamelEvent.ExchangeEvent)
event).getExchange().getExchangeId());
+ if (count != null) {
+ count.incrementAndGet();
+ }
+ }
+ }
+ });
+ context.getCamelContextExtension().setSendHook(new RandomHook());
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.addComponent("async", new MyAsyncComponent());
+
+ from("direct:stress")
+
.errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(1).redeliveryDelay(0))
+ .to("mock:s1")
+ .toD("mock:s${header.n}")
+
.recipientList(constant("mock:s4,async:bye:camel?delay=0"))
+ .routingSlip(constant("mock:s5,mock:s6"))
+ .to("mock:end");
+ }
+ };
+ }
+
+ @Test
+ void testEveryExchangeCompletesExactlyOnce() throws Exception {
+ List<CompletableFuture<Exchange>> futures = new ArrayList<>();
+ ExecutorService senders = Executors.newFixedThreadPool(SENDERS);
+ try {
+ List<Future<List<CompletableFuture<Exchange>>>> sent = new
ArrayList<>();
+ for (int t = 0; t < SENDERS; t++) {
+ sent.add(senders.submit(() -> {
+ List<CompletableFuture<Exchange>> mine = new ArrayList<>();
+ for (int i = 0; i < PER_SENDER; i++) {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setBody("Hello " + i);
+ exchange.getMessage().setHeader("n", 2 + (i % 2));
+ completions.put(exchange.getExchangeId(), new
AtomicInteger());
+ mine.add(template.asyncSend("direct:stress",
exchange));
+ }
+ return mine;
+ }));
+ }
+ for (Future<List<CompletableFuture<Exchange>>> f : sent) {
+ futures.addAll(f.get(60, TimeUnit.SECONDS));
+ }
+ } finally {
+ senders.shutdownNow();
+ }
+
+ CompletableFuture.allOf(futures.toArray(new
CompletableFuture[0])).get(60, TimeUnit.SECONDS);
+ // stop the context: every thread that could complete an exchange a
second time is done after this
+ context.stop();
+
+ assertEquals(SENDERS * PER_SENDER, completions.size());
+ completions.forEach((id, count) -> assertEquals(1, count.get(),
"completions of " + id));
+ assertEquals(before.get(), after.get());
+ // the hook did stop and delay sends, so those paths were exercised
+ assertTrue(stopped.get() > 0, "stopped " + stopped);
+ assertTrue(delayed.get() > 0, "delayed " + delayed);
+ }
+
+ private final class RandomHook implements SendHook {
+
+ @Override
+ public long beforeSend(Exchange exchange, Endpoint endpoint) {
+ before.incrementAndGet();
+ if (endpoint.getEndpointUri().startsWith("direct:")) {
+ // the send of the test into the route: it must reach the
route, so the exchange completes there
+ return PROCEED;
+ }
+ int dice = ThreadLocalRandom.current().nextInt(10);
+ switch (dice) {
+ case 0, 1:
+ stopped.incrementAndGet();
+ exchange.setException(new IllegalStateException("Stopped
by the send hook"));
+ return PROCEED;
+ case 2:
+ return SKIP;
+ case 3, 4:
+ delayed.incrementAndGet();
+ return 1 + ThreadLocalRandom.current().nextInt(3);
+ case 5:
+ exchange.getMessage().setBody("changed");
+ return PROCEED;
+ default:
+ return PROCEED;
+ }
+ }
+
+ @Override
+ public void afterSend(Exchange exchange, Endpoint endpoint) {
+ after.incrementAndGet();
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SendHookTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SendHookTest.java
new file mode 100644
index 000000000000..48faab98af3b
--- /dev/null
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/SendHookTest.java
@@ -0,0 +1,500 @@
+/*
+ * 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.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.processor.async.MyAsyncComponent;
+import org.apache.camel.processor.errorhandler.RedeliveryErrorHandler;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeSendingEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeSentEvent;
+import org.apache.camel.spi.SendHook;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.EventNotifierSupport;
+import org.apache.camel.support.PluginHelper;
+import org.apache.camel.support.SendHelper;
+import org.apache.camel.support.service.ServiceHelper;
+import org.junit.jupiter.api.Test;
+
+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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The {@link SendHook} runs around the sends to an endpoint, at each of the
places that send and fire the sending event
+ * (CAMEL-26331).
+ */
+class SendHookTest extends ContextTestSupport {
+
+ private static final String STOPPED = "Stopped by the send hook";
+
+ // the hook calls and the sending/sent events, in the order they happened
+ private final Queue<String> calls = new ConcurrentLinkedQueue<>();
+ private final TestHook hook = new TestHook();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ DefaultCamelContext context = (DefaultCamelContext)
super.createCamelContext();
+ context.getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) {
+ if (event instanceof ExchangeSendingEvent sending) {
+ calls.add("sending:" +
sending.getEndpoint().getEndpointUri());
+ } else if (event instanceof ExchangeSentEvent sent) {
+ calls.add("sent:" + sent.getEndpoint().getEndpointUri());
+ }
+ }
+ });
+ context.getCamelContextExtension().setSendHook(hook);
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.addComponent("async", new MyAsyncComponent());
+
+ from("direct:to").to("mock:a");
+ from("direct:toD").toD("mock:${header.target}");
+
from("direct:enrich").enrich("direct:resource").to("mock:enriched");
+ from("direct:resource").transform(constant("resource"));
+ from("direct:wireTap").wireTap("mock:tap").to("mock:afterTap");
+ from("direct:recipientList").recipientList(header("to"));
+ from("direct:routingSlip").routingSlip(header("slip"));
+ from("direct:multicast").multicast().to("mock:m1", "mock:m2");
+
from("direct:async").to("async:bye:camel").to("mock:afterAsync");
+
+ from("direct:change").to("mock:change");
+
from("direct:reply").to("mock:replyTarget").to("mock:afterReply");
+ from("direct:skip").to("mock:skipped").to("mock:afterSkip");
+ from("direct:delay").to("mock:delayed").to("mock:afterDelay");
+
+
from("direct:stopTo").errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0))
+ .to("mock:stopped");
+ from("direct:stopRecipients")
+
.errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0))
+ .recipientList(constant("mock:ok,mock:stopped"));
+
from("direct:stopSlip").errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2).redeliveryDelay(0))
+ .routingSlip(constant("mock:ok,mock:stopped"));
+ from("direct:stopOnException")
+
.onException(IllegalStateException.class).handled(true).to("mock:handled").end()
+ .to("mock:stopped");
+ from("direct:stopTry")
+
.doTry().recipientList(constant("mock:stopped")).end().doCatch(IllegalStateException.class)
+ .to("mock:caught").end();
+ }
+ };
+ }
+
+ @Test
+ void testTo() throws Exception {
+ getMockEndpoint("mock:a").expectedBodiesReceived("Hello");
+ template.sendBody("direct:to", "Hello");
+ assertMockEndpointsSatisfied();
+ assertHookAroundSend("mock://a");
+ }
+
+ @Test
+ void testToD() throws Exception {
+ getMockEndpoint("mock:dyn").expectedBodiesReceived("Hello");
+ template.sendBodyAndHeader("direct:toD", "Hello", "target", "dyn");
+ assertMockEndpointsSatisfied();
+ assertHookAroundSend("mock://dyn");
+ }
+
+ @Test
+ void testEnrich() throws Exception {
+ getMockEndpoint("mock:enriched").expectedBodiesReceived("resource");
+ template.sendBody("direct:enrich", "Hello");
+ assertMockEndpointsSatisfied();
+ assertHookAroundSend("direct://resource");
+ assertHookAroundSend("mock://enriched");
+ }
+
+ @Test
+ void testWireTap() throws Exception {
+ getMockEndpoint("mock:tap").expectedBodiesReceived("Hello");
+ getMockEndpoint("mock:afterTap").expectedBodiesReceived("Hello");
+ template.sendBody("direct:wireTap", "Hello");
+ assertMockEndpointsSatisfied();
+ // the tap is sent on another thread, and its after hook runs after
the mock got the message
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
calls.contains("sent:mock://tap"));
+ assertHookAroundSend("mock://tap");
+ assertHookAroundSend("mock://afterTap");
+ }
+
+ @Test
+ void testRecipientList() throws Exception {
+ getMockEndpoint("mock:r1").expectedBodiesReceived("Hello");
+ getMockEndpoint("mock:r2").expectedBodiesReceived("Hello");
+ template.sendBodyAndHeader("direct:recipientList", "Hello", "to",
"mock:r1,mock:r2");
+ assertMockEndpointsSatisfied();
+ assertHookAroundSend("mock://r1");
+ assertHookAroundSend("mock://r2");
+ }
+
+ @Test
+ void testRoutingSlip() throws Exception {
+ getMockEndpoint("mock:s1").expectedBodiesReceived("Hello");
+ getMockEndpoint("mock:s2").expectedBodiesReceived("Hello");
+ template.sendBodyAndHeader("direct:routingSlip", "Hello", "slip",
"mock:s1,mock:s2");
+ assertMockEndpointsSatisfied();
+ assertHookAroundSend("mock://s1");
+ assertHookAroundSend("mock://s2");
+ }
+
+ @Test
+ void testMulticast() throws Exception {
+ getMockEndpoint("mock:m1").expectedBodiesReceived("Hello");
+ getMockEndpoint("mock:m2").expectedBodiesReceived("Hello");
+ template.sendBody("direct:multicast", "Hello");
+ assertMockEndpointsSatisfied();
+ assertHookAroundSend("mock://m1");
+ assertHookAroundSend("mock://m2");
+ }
+
+ @Test
+ void testProducerTemplate() throws Exception {
+ getMockEndpoint("mock:pt").expectedBodiesReceived("Hello");
+ template.sendBody("mock:pt", "Hello");
+ assertMockEndpointsSatisfied();
+ assertHookAroundSend("mock://pt");
+ }
+
+ @Test
+ void testProducerTemplateAsync() throws Exception {
+ getMockEndpoint("mock:ptAsync").expectedBodiesReceived("Hello");
+ template.asyncSendBody("mock:ptAsync", "Hello").get(10,
TimeUnit.SECONDS);
+ assertMockEndpointsSatisfied();
+ // the hook runs once, though this path fires the sending and the sent
event twice (as it did before the hook)
+ assertEquals(List.of("sending:mock://ptAsync",
"sending:mock://ptAsync", "before:mock://ptAsync",
+ "after:mock://ptAsync", "sent:mock://ptAsync",
"sent:mock://ptAsync"), calls("mock://ptAsync"));
+ }
+
+ @Test
+ void testDeferredProducer() throws Exception {
+ getMockEndpoint("mock:deferred").expectedBodiesReceived("Hello");
+ Producer producer =
PluginHelper.getDeferServiceFactory(context).createProducer(context.getEndpoint("mock:deferred"));
+ ServiceHelper.startService(producer);
+ try {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setBody("Hello");
+ producer.process(exchange);
+ assertNull(exchange.getException());
+ } finally {
+ ServiceHelper.stopService(producer);
+ }
+ assertMockEndpointsSatisfied();
+ assertHookAroundSend("mock://deferred");
+ }
+
+ @Test
+ void testAsyncProducer() throws Exception {
+ getMockEndpoint("mock:afterAsync").expectedBodiesReceived("Bye Camel");
+ Object reply = template.requestBody("direct:async", "Hello");
+ assertEquals("Bye Camel", reply);
+ assertMockEndpointsSatisfied();
+ assertHookAroundSend("async://bye:camel");
+ }
+
+ @Test
+ void testChangedMessageReachesProducer() throws Exception {
+ getMockEndpoint("mock:change").expectedBodiesReceived("changed");
+ template.sendBody("direct:change", "Hello");
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ void testChangedReplyReachesRoute() throws Exception {
+ getMockEndpoint("mock:replyTarget").expectedBodiesReceived("Hello");
+ getMockEndpoint("mock:afterReply").expectedBodiesReceived("reply
changed");
+ Object reply = template.requestBody("direct:reply", "Hello");
+ assertEquals("reply changed", reply);
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ void testSkip() throws Exception {
+ getMockEndpoint("mock:skipped").expectedMessageCount(0);
+ getMockEndpoint("mock:afterSkip").expectedBodiesReceived("Hello");
+ Exchange out = template.send("direct:skip", e ->
e.getMessage().setBody("Hello"));
+ assertNull(out.getException());
+ assertMockEndpointsSatisfied();
+ // the skipped send has its after hook and its events
+ assertHookAroundSend("mock://skipped");
+ }
+
+ @Test
+ void testDelayWaitsOnTheCallerThread() throws Exception {
+ MockEndpoint delayed = getMockEndpoint("mock:delayed");
+ delayed.expectedBodiesReceived("Hello");
+ getMockEndpoint("mock:afterDelay").expectedBodiesReceived("Hello");
+ List<Thread> threads = new CopyOnWriteArrayList<>();
+ delayed.whenAnyExchangeReceived(e ->
threads.add(Thread.currentThread()));
+
+ template.sendBody("direct:delay", "Hello");
+
+ assertMockEndpointsSatisfied();
+ assertHookAroundSend("mock://delayed");
+ // the thread that sent waited for the delay, as for a slow endpoint,
and then called the producer
+ assertEquals(List.of(Thread.currentThread()), threads);
+ }
+
+ @Test
+ void testRecipientListWithoutHookCallsTheProducerItself() throws Exception
{
+ MockEndpoint rl = getMockEndpoint("mock:rl");
+ rl.expectedMessageCount(3);
+ List<String> stack = new CopyOnWriteArrayList<>();
+ rl.whenAnyExchangeReceived(e ->
stack.addAll(callersUpToTheErrorHandler()));
+ String hooked = SendHelper.class.getName() + "$SendHookProducer";
+
+ // no hook: the error handler of the recipient calls the producer
itself, as when there is no send hook at all
+ context.getCamelContextExtension().setSendHook(null);
+ template.sendBodyAndHeader("direct:recipientList", "Hello", "to",
"mock:rl");
+ assertTrue(stack.get(stack.size() -
1).startsWith(RedeliveryErrorHandler.class.getName()), stack::toString);
+ assertFalse(stack.contains(hooked), stack::toString);
+
+ // a hook installed later runs for the same recipient, through another
error handler
+ stack.clear();
+ context.getCamelContextExtension().setSendHook(hook);
+ template.sendBodyAndHeader("direct:recipientList", "Hello", "to",
"mock:rl");
+ assertTrue(stack.contains(hooked), stack::toString);
+ assertEquals(1, hook.before("mock://rl"));
+
+ // and when it is removed again, the producer is called by itself again
+ stack.clear();
+ context.getCamelContextExtension().setSendHook(null);
+ template.sendBodyAndHeader("direct:recipientList", "Hello", "to",
"mock:rl");
+ assertFalse(stack.contains(hooked), stack::toString);
+ assertEquals(1, hook.before("mock://rl"));
+ assertMockEndpointsSatisfied();
+ }
+
+ /**
+ * The classes on the stack from the producer up to the (first) error
handler that called it.
+ */
+ private static List<String> callersUpToTheErrorHandler() {
+ List<String> classes = StackWalker.getInstance()
+ .walk(frames ->
frames.map(StackWalker.StackFrame::getClassName).toList());
+ for (int i = 0; i < classes.size(); i++) {
+ if
(classes.get(i).startsWith(RedeliveryErrorHandler.class.getName())) {
+ return classes.subList(0, i + 1);
+ }
+ }
+ return classes;
+ }
+
+ @Test
+ void testStoppedSendIsRedeliveredAndDeadLettered() throws Exception {
+ getMockEndpoint("mock:stopped").expectedMessageCount(0);
+ MockEndpoint dead = getMockEndpoint("mock:dead");
+ dead.expectedMessageCount(1);
+
+ template.sendBody("direct:stopTo", "Hello");
+
+ assertMockEndpointsSatisfied();
+ assertStoppedAndDeadLettered(dead);
+ // the first attempt and 2 redeliveries: the hook stopped each of them
+ assertEquals(3, hook.before("mock://stopped"));
+ assertEquals(3, hook.after("mock://stopped"));
+ }
+
+ @Test
+ void testStoppedSendInRecipientListIsRedeliveredPerRecipient() throws
Exception {
+ // the hook runs inside the error handler of the recipient, so only
the stopped recipient is redelivered
+ getMockEndpoint("mock:ok").expectedMessageCount(1);
+ getMockEndpoint("mock:stopped").expectedMessageCount(0);
+ MockEndpoint dead = getMockEndpoint("mock:dead");
+ dead.expectedMessageCount(1);
+
+ template.sendBody("direct:stopRecipients", "Hello");
+
+ assertMockEndpointsSatisfied();
+ assertStoppedAndDeadLettered(dead);
+ assertEquals(3, hook.before("mock://stopped"));
+ assertEquals(1, hook.before("mock://ok"));
+ }
+
+ @Test
+ void testStoppedSendInRoutingSlipIsRedeliveredPerStep() throws Exception {
+ // the hook runs inside the error handler of the step, so only the
stopped step is redelivered
+ getMockEndpoint("mock:ok").expectedMessageCount(1);
+ getMockEndpoint("mock:stopped").expectedMessageCount(0);
+ MockEndpoint dead = getMockEndpoint("mock:dead");
+ dead.expectedMessageCount(1);
+
+ template.sendBody("direct:stopSlip", "Hello");
+
+ assertMockEndpointsSatisfied();
+ assertStoppedAndDeadLettered(dead);
+ assertEquals(3, hook.before("mock://stopped"));
+ assertEquals(1, hook.before("mock://ok"));
+ }
+
+ @Test
+ void testStoppedSendGoesToOnException() throws Exception {
+ getMockEndpoint("mock:stopped").expectedMessageCount(0);
+ MockEndpoint handled = getMockEndpoint("mock:handled");
+ handled.expectedMessageCount(1);
+
+ Exchange out = template.send("direct:stopOnException", e ->
e.getMessage().setBody("Hello"));
+
+ assertNull(out.getException());
+ assertMockEndpointsSatisfied();
+ Exception caught =
handled.getReceivedExchanges().get(0).getProperty(Exchange.EXCEPTION_CAUGHT,
Exception.class);
+ assertInstanceOf(IllegalStateException.class, caught);
+ assertEquals(STOPPED, caught.getMessage());
+ }
+
+ @Test
+ void testStoppedSendGoesToDoCatch() throws Exception {
+ getMockEndpoint("mock:stopped").expectedMessageCount(0);
+ getMockEndpoint("mock:caught").expectedMessageCount(1);
+
+ Exchange out = template.send("direct:stopTry", e ->
e.getMessage().setBody("Hello"));
+
+ assertNull(out.getException());
+ assertMockEndpointsSatisfied();
+ assertEquals(1, hook.before("mock://stopped"));
+ }
+
+ @Test
+ void testStoppedSendFromProducerTemplate() throws Exception {
+ getMockEndpoint("mock:stopped").expectedMessageCount(0);
+
+ Exchange out = template.send("mock:stopped", e ->
e.getMessage().setBody("Hello"));
+
+ assertMockEndpointsSatisfied();
+ assertInstanceOf(IllegalStateException.class, out.getException());
+ assertEquals(STOPPED, out.getException().getMessage());
+ assertHookAroundSend("mock://stopped");
+ }
+
+ @Test
+ void testEventsAreTheSameWithoutHook() throws Exception {
+ List<String> withHook = runAllPaths();
+ // the hook runs between the sending and the sent event of its send
+ assertEquals(List.of("sending:mock://a", "before:mock://a",
"after:mock://a", "sent:mock://a"),
+ calls("mock://a"));
+
+ context.getCamelContextExtension().setSendHook(null);
+ assertNull(context.getCamelContextExtension().getSendHook());
+ List<String> withoutHook = runAllPaths();
+
+ // the hook adds its calls, and nothing else changes in the events
+ List<String> withHookEventsOnly = withHook.stream().filter(c ->
c.startsWith("sen")).toList();
+ assertEquals(withHookEventsOnly, withoutHook);
+ assertTrue(withoutHook.stream().allMatch(c -> c.startsWith("sen")),
withoutHook::toString);
+ }
+
+ private List<String> runAllPaths() throws Exception {
+ calls.clear();
+ template.sendBody("direct:to", "Hello");
+ template.sendBodyAndHeader("direct:toD", "Hello", "target", "dyn");
+ template.sendBody("direct:enrich", "Hello");
+ template.sendBodyAndHeader("direct:recipientList", "Hello", "to",
"mock:r1,mock:r2");
+ template.sendBodyAndHeader("direct:routingSlip", "Hello", "slip",
"mock:s1,mock:s2");
+ template.sendBody("direct:multicast", "Hello");
+ template.sendBody("mock:pt", "Hello");
+ template.asyncSendBody("mock:ptAsync", "Hello").get(10,
TimeUnit.SECONDS);
+ template.requestBody("direct:async", "Hello");
+ return new ArrayList<>(calls);
+ }
+
+ private void assertStoppedAndDeadLettered(MockEndpoint dead) {
+ Exchange deadExchange = dead.getReceivedExchanges().get(0);
+ Exception caught = deadExchange.getProperty(Exchange.EXCEPTION_CAUGHT,
Exception.class);
+ assertNotNull(caught);
+ assertInstanceOf(IllegalStateException.class, caught);
+ assertEquals(STOPPED, caught.getMessage());
+ }
+
+ /**
+ * Asserts that each send to the endpoint ran the hook once before and
once after, between the sending and the sent
+ * event of that send.
+ */
+ private void assertHookAroundSend(String uri) {
+ List<String> list = calls(uri);
+ assertEquals(List.of("sending:" + uri, "before:" + uri, "after:" +
uri, "sent:" + uri), list, list::toString);
+ assertEquals(1, hook.before(uri));
+ assertEquals(1, hook.after(uri));
+ }
+
+ private List<String> calls(String uri) {
+ return calls.stream().filter(c -> c.endsWith(":" + uri)).toList();
+ }
+
+ private final class TestHook implements SendHook {
+
+ @Override
+ public long beforeSend(Exchange exchange, Endpoint endpoint) {
+ String uri = endpoint.getEndpointUri();
+ calls.add("before:" + uri);
+ switch (uri) {
+ case "mock://stopped":
+ exchange.setException(new IllegalStateException(STOPPED));
+ return PROCEED;
+ case "mock://change":
+ exchange.getMessage().setBody("changed");
+ return PROCEED;
+ case "mock://skipped":
+ return SKIP;
+ case "mock://delayed":
+ return 50;
+ default:
+ return PROCEED;
+ }
+ }
+
+ @Override
+ public void afterSend(Exchange exchange, Endpoint endpoint) {
+ String uri = endpoint.getEndpointUri();
+ calls.add("after:" + uri);
+ if ("mock://replyTarget".equals(uri)) {
+ exchange.getMessage().setBody("reply changed");
+ }
+ }
+
+ int before(String uri) {
+ return (int) calls.stream().filter(c -> c.equals("before:" +
uri)).count();
+ }
+
+ int after(String uri) {
+ return (int) calls.stream().filter(c -> c.equals("after:" +
uri)).count();
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/support/SendHelperTest.java
b/core/camel-core/src/test/java/org/apache/camel/support/SendHelperTest.java
new file mode 100644
index 000000000000..12e14d8c9fc7
--- /dev/null
+++ b/core/camel-core/src/test/java/org/apache/camel/support/SendHelperTest.java
@@ -0,0 +1,544 @@
+/*
+ * 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.ArrayList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProducer;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.SendHook;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.util.TimeSource;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Isolated;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+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 callback contract of
+ * {@link SendHelper#process(Exchange, Endpoint,
org.apache.camel.AsyncProcessor, AsyncCallback)} with a
+ * {@link SendHook}: the callback is called exactly once on every path, with
the value that process returned.
+ * <p/>
+ * A test installs a virtual clock in {@link TimeSource}, which is JVM-wide,
so these tests run isolated from the other
+ * tests.
+ */
+@Isolated
+class SendHelperTest extends ContextTestSupport {
+
+ private static final String MODE = "hookMode";
+ private static final String KIND = "producerKind";
+ private static final String AFTER_COUNT = "afterCount";
+ private static final String PRODUCER_COUNT = "producerCount";
+ private static final String FIXED_DELAY = "fixedDelay";
+
+ private ExecutorService producerThreads;
+ private Endpoint endpoint;
+
+ enum Mode {
+ PROCEED,
+ CHANGE,
+ STOP,
+ SKIP,
+ DELAY,
+ RESERVED,
+ BEFORE_THROWS,
+ AFTER_THROWS
+ }
+
+ enum Kind {
+ SYNC,
+ ASYNC,
+ THROWS
+ }
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ super.setUp();
+ producerThreads = Executors.newFixedThreadPool(4);
+ endpoint = context.getEndpoint("mock:target");
+ context.getCamelContextExtension().setSendHook(new ModeHook());
+ }
+
+ @Override
+ @AfterEach
+ public void tearDown() throws Exception {
+ producerThreads.shutdownNow();
+ super.tearDown();
+ }
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @ParameterizedTest
+ @EnumSource(Mode.class)
+ void testSyncProducer(Mode mode) throws Exception {
+ assertOneSend(mode, Kind.SYNC, false);
+ }
+
+ @ParameterizedTest
+ @EnumSource(Mode.class)
+ void testAsyncProducer(Mode mode) throws Exception {
+ assertOneSend(mode, Kind.ASYNC, false);
+ }
+
+ @ParameterizedTest
+ @EnumSource(Mode.class)
+ void testProducerThrows(Mode mode) throws Exception {
+ assertOneSend(mode, Kind.THROWS, false);
+ }
+
+ @ParameterizedTest
+ @EnumSource(Mode.class)
+ void testTransacted(Mode mode) throws Exception {
+ assertOneSend(mode, Kind.SYNC, true);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = { false, true })
+ void testDelayWaitsOnTheCallerThreadThroughTimeSource(boolean transacted) {
+ Exchange exchange = newExchange(Mode.DELAY, Kind.SYNC);
+ exchange.setProperty(FIXED_DELAY, 2500L);
+ exchange.getExchangeExtension().setTransacted(transacted);
+ List<Thread> producerThread = new ArrayList<>();
+ Completion completion = new Completion();
+ VirtualClock clock = new VirtualClock();
+
+ TimeSource.Provider previous = TimeSource.install(clock);
+ boolean sync;
+ try {
+ sync = SendHelper.process(exchange, endpoint, new
AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback
callback) {
+ producerThread.add(Thread.currentThread());
+ callback.done(true);
+ return true;
+ }
+ }, completion);
+ } finally {
+ TimeSource.install(previous);
+ }
+
+ // this thread waited for the delay through TimeSource (in virtual
time here), and then called the producer
+ assertEquals(List.of(Thread.currentThread()), clock.sleepers);
+ assertEquals(List.of(2500L), clock.sleeps);
+ assertEquals(List.of(Thread.currentThread()), producerThread);
+ assertTrue(sync);
+ assertEquals(1, completion.count.get());
+ assertTrue(completion.doneSync);
+ assertEquals(1, count(exchange, AFTER_COUNT));
+ assertNull(exchange.getException());
+ }
+
+ @Test
+ void testDelayInterrupted() {
+ Exchange exchange = newExchange(Mode.DELAY, Kind.SYNC);
+ Completion completion = new Completion();
+
+ // the thread is interrupted, so the wait for the delay ends right away
+ Thread.currentThread().interrupt();
+ boolean sync;
+ boolean interrupted;
+ try {
+ sync = SendHelper.process(exchange, endpoint, new TestProducer(),
completion);
+ } finally {
+ // the helper keeps the interrupted status of the thread; clear it
for the next tests
+ interrupted = Thread.interrupted();
+ }
+
+ // the exchange is marked as interrupted, so Camel stops routing it
+ assertTrue(interrupted);
+ assertTrue(sync);
+ assertEquals(1, completion.count.get());
+ assertTrue(completion.doneSync);
+ assertInstanceOf(InterruptedException.class, exchange.getException());
+ assertTrue(exchange.getExchangeExtension().isInterrupted());
+ assertEquals(0, count(exchange, PRODUCER_COUNT));
+ assertEquals(1, count(exchange, AFTER_COUNT));
+ }
+
+ @Test
+ void testReservedValueFailsTheSend() {
+ Exchange exchange = newExchange(Mode.RESERVED, Kind.SYNC);
+ Completion completion = new Completion();
+
+ boolean sync = SendHelper.process(exchange, endpoint, new
TestProducer(), completion);
+
+ assertTrue(sync);
+ assertEquals(1, completion.count.get());
+ assertTrue(completion.doneSync);
+ assertInstanceOf(IllegalArgumentException.class,
exchange.getException());
+ assertEquals("SendHook.beforeSend returned -2: negative values other
than SKIP are reserved",
+ exchange.getException().getMessage());
+ assertEquals(0, count(exchange, PRODUCER_COUNT));
+ assertEquals(1, count(exchange, AFTER_COUNT));
+ }
+
+ @Test
+ void testSendHookProducerPassesTheLifecycleOn() throws Exception {
+ AsyncProducer producer = endpoint.createAsyncProducer();
+ AsyncProducer hooked = SendHelper.newSendHookProducer(producer);
+
+ ServiceHelper.startService(hooked);
+ assertTrue(ServiceHelper.isStarted(producer));
+
+ // as an error handler does with its output when it is shut down
+ ServiceHelper.stopAndShutdownServices(hooked);
+ assertTrue(ServiceHelper.isStopped(producer));
+ assertTrue(((ServiceSupport) producer).isShutdown());
+ }
+
+ @Test
+ void testNoHookCallsTheProducerOnly() {
+ context.getCamelContextExtension().setSendHook(null);
+ Exchange exchange = newExchange(Mode.STOP, Kind.SYNC);
+ Completion completion = new Completion();
+
+ boolean sync = SendHelper.process(exchange, endpoint, new
TestProducer(), completion);
+
+ assertTrue(sync);
+ assertEquals(1, completion.count.get());
+ assertEquals(1, count(exchange, PRODUCER_COUNT));
+ assertEquals(0, count(exchange, AFTER_COUNT));
+ assertNull(exchange.getException());
+ }
+
+ @Test
+ void testExchangeThatHasFailedIsSentAsWithoutHook() {
+ Exchange exchange = newExchange(Mode.PROCEED, Kind.SYNC);
+ IllegalArgumentException failure = new
IllegalArgumentException("Failed before the send");
+ exchange.setException(failure);
+ Completion completion = new Completion();
+
+ boolean sync = SendHelper.process(exchange, endpoint, new
TestProducer(), completion);
+
+ // the exception was there before the hook, so the hook did not stop
the send
+ assertTrue(sync);
+ assertEquals(1, completion.count.get());
+ assertEquals(1, count(exchange, PRODUCER_COUNT));
+ assertEquals(1, count(exchange, AFTER_COUNT));
+ assertSame(failure, exchange.getException());
+ }
+
+ /**
+ * Many concurrent sends, each with a random hook outcome and a random
producer: every send completes exactly once,
+ * with the value process returned, and the after hook runs once for each
before hook that returned.
+ */
+ @Test
+ void testStressExactlyOnce() throws Exception {
+ int threads = 16;
+ int perThread = 500;
+ List<Send> sends = new ArrayList<>();
+ ExecutorService senders = Executors.newFixedThreadPool(threads);
+ try {
+ List<Future<List<Send>>> futures = new ArrayList<>();
+ for (int t = 0; t < threads; t++) {
+ futures.add(senders.submit(() -> {
+ List<Send> mine = new ArrayList<>();
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ for (int i = 0; i < perThread; i++) {
+ Mode mode =
Mode.values()[random.nextInt(Mode.values().length)];
+ Kind kind =
Kind.values()[random.nextInt(Kind.values().length)];
+ Send send = new Send(newExchange(mode, kind), mode,
kind);
+ send.start(endpoint, new TestProducer());
+ mine.add(send);
+ }
+ return mine;
+ }));
+ }
+ for (Future<List<Send>> future : futures) {
+ sends.addAll(future.get(60, TimeUnit.SECONDS));
+ }
+ } finally {
+ senders.shutdownNow();
+ }
+
+ for (Send send : sends) {
+ assertTrue(send.completion.latch.await(20, TimeUnit.SECONDS),
"send not completed: " + send);
+ }
+ // stop the threads that complete sends, so a second completion would
have happened by now
+ producerThreads.shutdown();
+ assertTrue(producerThreads.awaitTermination(20, TimeUnit.SECONDS));
+ context.stop();
+
+ assertEquals(threads * perThread, sends.size());
+ for (Send send : sends) {
+ send.assertCompletedOnce();
+ }
+ }
+
+ private void assertOneSend(Mode mode, Kind kind, boolean transacted)
throws Exception {
+ Exchange exchange = newExchange(mode, kind);
+ exchange.getExchangeExtension().setTransacted(transacted);
+ Send send = new Send(exchange, mode, kind);
+ send.start(endpoint, new TestProducer());
+ assertTrue(send.completion.latch.await(20, TimeUnit.SECONDS));
+ producerThreads.shutdown();
+ assertTrue(producerThreads.awaitTermination(20, TimeUnit.SECONDS));
+ context.stop();
+ send.assertCompletedOnce();
+ if (transacted) {
+ // a transacted exchange (with a synchronous producer) completes
on the thread that sent it
+ assertTrue(send.returned == null || send.returned, send::toString);
+ }
+ }
+
+ private Exchange newExchange(Mode mode, Kind kind) {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setBody("Hello");
+ exchange.setProperty(MODE, mode);
+ exchange.setProperty(KIND, kind);
+ exchange.setProperty(AFTER_COUNT, new AtomicInteger());
+ exchange.setProperty(PRODUCER_COUNT, new AtomicInteger());
+ return exchange;
+ }
+
+ private static int count(Exchange exchange, String name) {
+ return exchange.getProperty(name, AtomicInteger.class).get();
+ }
+
+ /**
+ * One call of SendHelper.process, done the way the callers of a producer
do it.
+ */
+ private static final class Send {
+ private final Exchange exchange;
+ private final Mode mode;
+ private final Kind kind;
+ private final Completion completion = new Completion();
+ // null when process threw
+ private volatile Boolean returned;
+ private volatile int countWhenReturned;
+ private volatile RuntimeException thrown;
+
+ private Send(Exchange exchange, Mode mode, Kind kind) {
+ this.exchange = exchange;
+ this.mode = mode;
+ this.kind = kind;
+ }
+
+ void start(Endpoint endpoint, TestProducer producer) {
+ try {
+ boolean sync = SendHelper.process(exchange, endpoint,
producer, completion);
+ countWhenReturned = completion.count.get();
+ returned = sync;
+ } catch (RuntimeException e) {
+ // the producer threw: the callers then set the exception and
complete the send themselves
+ thrown = e;
+ countWhenReturned = completion.count.get();
+ exchange.setException(e);
+ completion.done(true);
+ }
+ }
+
+ void assertCompletedOnce() {
+ assertEquals(1, completion.count.get(), this::toString);
+ boolean producerCalled = count(exchange, PRODUCER_COUNT) == 1;
+ assertTrue(count(exchange, PRODUCER_COUNT) <= 1, this::toString);
+ boolean stoppedByHook
+ = mode == Mode.STOP || mode == Mode.SKIP || mode ==
Mode.RESERVED || mode == Mode.BEFORE_THROWS;
+ assertEquals(!stoppedByHook, producerCalled, this::toString);
+ // the after hook runs once for every before hook that returned
+ assertEquals(mode == Mode.BEFORE_THROWS ? 0 : 1, count(exchange,
AFTER_COUNT), this::toString);
+
+ if (thrown != null) {
+ // only a producer that throws on the caller thread is thrown
to the caller, and the callback was not
+ // called before
+ assertEquals(Kind.THROWS, kind, this::toString);
+ assertEquals(0, countWhenReturned, this::toString);
+ assertSame(thrown, exchange.getException(), this::toString);
+ return;
+ }
+ // process returned the value the callback was called with, and a
send that returned true was complete
+ assertEquals(returned, completion.doneSync, this::toString);
+ if (returned) {
+ assertEquals(1, countWhenReturned, this::toString);
+ }
+
+ switch (mode) {
+ case STOP -> assertEquals("Stopped by the hook",
exchange.getException().getMessage());
+ case RESERVED ->
assertInstanceOf(IllegalArgumentException.class, exchange.getException(),
this::toString);
+ case BEFORE_THROWS -> assertEquals("Before failed",
exchange.getException().getMessage());
+ case AFTER_THROWS -> {
+ assertNotNull(exchange.getException(), this::toString);
+ if (kind == Kind.THROWS) {
+ assertEquals("Producer failed",
exchange.getException().getMessage());
+ assertEquals("After failed",
exchange.getException().getSuppressed()[0].getMessage());
+ } else {
+ assertEquals("After failed",
exchange.getException().getMessage());
+ }
+ }
+ case SKIP -> {
+ assertNull(exchange.getException(), this::toString);
+ assertEquals("Hello", exchange.getMessage().getBody());
+ }
+ default -> {
+ if (kind == Kind.THROWS) {
+ assertEquals("Producer failed",
exchange.getException().getMessage(), this::toString);
+ } else {
+ assertNull(exchange.getException(), this::toString);
+ String expected = mode == Mode.CHANGE ? "Changed
reply" : "Reply";
+ assertEquals(expected,
exchange.getMessage().getBody(), this::toString);
+ }
+ }
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "Send[mode=" + mode + ", kind=" + kind + ", transacted=" +
exchange.isTransacted() + ", returned="
+ + returned + ", thrown=" + thrown + ", calls=" +
completion.count + ", doneSync=" + completion.doneSync
+ + ", exception=" + exchange.getException() + "]";
+ }
+ }
+
+ private static final class Completion implements AsyncCallback {
+ private final AtomicInteger count = new AtomicInteger();
+ private final CountDownLatch latch = new CountDownLatch(1);
+ private volatile boolean doneSync;
+
+ @Override
+ public void done(boolean doneSync) {
+ this.doneSync = doneSync;
+ count.incrementAndGet();
+ latch.countDown();
+ }
+ }
+
+ /**
+ * A producer that is synchronous, asynchronous (completes on another
thread) or throws, by the exchange.
+ */
+ private final class TestProducer extends AsyncProcessorSupport {
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ exchange.getProperty(PRODUCER_COUNT,
AtomicInteger.class).incrementAndGet();
+ Kind kind = exchange.getProperty(KIND, Kind.class);
+ switch (kind) {
+ case SYNC:
+ reply(exchange);
+ callback.done(true);
+ return true;
+ case ASYNC:
+ producerThreads.execute(() -> {
+ reply(exchange);
+ callback.done(false);
+ });
+ return false;
+ default:
+ throw new IllegalArgumentException("Producer failed");
+ }
+ }
+
+ private void reply(Exchange exchange) {
+ if ("Changed".equals(exchange.getMessage().getBody())) {
+ exchange.getMessage().setBody("Changed reply");
+ } else {
+ exchange.getMessage().setBody("Reply");
+ }
+ }
+ }
+
+ /**
+ * A clock that records the sleeps of the test thread, and the threads
that sleep, without waiting. Any other thread
+ * waits as usual.
+ */
+ private static final class VirtualClock implements TimeSource.Provider {
+ private final Thread testThread = Thread.currentThread();
+ private final List<Long> sleeps = new CopyOnWriteArrayList<>();
+ private final List<Thread> sleepers = new CopyOnWriteArrayList<>();
+
+ @Override
+ public long currentTimeMillis() {
+ return System.currentTimeMillis();
+ }
+
+ @Override
+ public long nanoTime() {
+ return System.nanoTime();
+ }
+
+ @Override
+ public void sleep(long millis) throws InterruptedException {
+ if (Thread.currentThread() != testThread) {
+ new CountDownLatch(1).await(millis, TimeUnit.MILLISECONDS);
+ return;
+ }
+ sleeps.add(millis);
+ sleepers.add(Thread.currentThread());
+ }
+ }
+
+ private static final class ModeHook implements SendHook {
+
+ @Override
+ public long beforeSend(Exchange exchange, Endpoint endpoint) {
+ Mode mode = exchange.getProperty(MODE, Mode.class);
+ switch (mode) {
+ case CHANGE:
+ exchange.getMessage().setBody("Changed");
+ return PROCEED;
+ case STOP:
+ exchange.setException(new IllegalStateException("Stopped
by the hook"));
+ return PROCEED;
+ case SKIP:
+ return SKIP;
+ case DELAY:
+ Long fixed = exchange.getProperty(FIXED_DELAY, Long.class);
+ return fixed != null ? fixed : 1 +
ThreadLocalRandom.current().nextInt(3);
+ case RESERVED:
+ return -2;
+ case BEFORE_THROWS:
+ throw new IllegalStateException("Before failed");
+ default:
+ return PROCEED;
+ }
+ }
+
+ @Override
+ public void afterSend(Exchange exchange, Endpoint endpoint) {
+ exchange.getProperty(AFTER_COUNT,
AtomicInteger.class).incrementAndGet();
+ if (exchange.getProperty(MODE, Mode.class) == Mode.AFTER_THROWS) {
+ throw new IllegalStateException("After failed");
+ }
+ }
+ }
+}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/SendHelper.java
b/core/camel-support/src/main/java/org/apache/camel/support/SendHelper.java
new file mode 100644
index 000000000000..8964c0dcc63a
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/SendHelper.java
@@ -0,0 +1,348 @@
+/*
+ * 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.io.IOException;
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.AsyncProducer;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.ShutdownableService;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.SendHook;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.util.StopWatch;
+import org.apache.camel.util.TimeSource;
+
+/**
+ * Helper for the places in Camel that send an {@link Exchange} to the
producer of an {@link Endpoint}: the
+ * {@code SendProcessor}, the {@link org.apache.camel.spi.ProducerCache} (used
by {@code toD}, {@code enrich},
+ * {@code wireTap}, {@code routingSlip} and the {@link
org.apache.camel.ProducerTemplate}), the recipient list and the
+ * deferred producers.
+ * <p/>
+ * It fires the {@link CamelEvent.ExchangeSendingEvent} before the send and
the {@link CamelEvent.ExchangeSentEvent},
+ * with the time taken, after it ({@link #notifySending(Exchange, Endpoint)}
and
+ * {@link #notifySent(Exchange, Endpoint, StopWatch)}), and calls the producer
with the {@link SendHook} of the
+ * {@link CamelContext}, if one is installed, around it
+ * ({@link #process(Exchange, Endpoint, AsyncProcessor, AsyncCallback)}).
+ * {@link #send(Exchange, Endpoint, AsyncProcessor, AsyncCallback)} does both.
+ * <p/>
+ * The events and the hook are apart because some places fire the events once
per send and call the producer inside an
+ * error handler that can call it again: the recipient list and the routing
slip. The hook runs where the producer is
+ * called, so it runs inside that error handler, for every attempt.
+ *
+ * @since 4.24
+ */
+public final class SendHelper {
+
+ private SendHelper() {
+ }
+
+ /**
+ * Fires the {@link CamelEvent.ExchangeSendingEvent}.
+ *
+ * @param exchange the exchange that is sent
+ * @param endpoint the endpoint it is sent to
+ * @return a started {@link StopWatch} when an {@link
CamelEvent.ExchangeSentEvent} must be fired when the
+ * send is done (see {@link #notifySent(Exchange,
Endpoint, StopWatch)}), or {@code null}
+ */
+ public static StopWatch notifySending(Exchange exchange, Endpoint
endpoint) {
+ if (EventHelper.notifyExchangeSending(exchange.getContext(), exchange,
endpoint)) {
+ return new StopWatch();
+ }
+ return null;
+ }
+
+ /**
+ * Fires the {@link CamelEvent.ExchangeSentEvent} with the time taken since
+ * {@link #notifySending(Exchange, Endpoint)}, when it returned a {@link
StopWatch}.
+ *
+ * @param exchange the exchange that was sent
+ * @param endpoint the endpoint it was sent to
+ * @param watch the watch {@link #notifySending(Exchange, Endpoint)}
returned, or {@code null} to fire nothing
+ */
+ public static void notifySent(Exchange exchange, Endpoint endpoint,
StopWatch watch) {
+ if (watch != null) {
+ long timeTaken = watch.taken();
+ EventHelper.notifyExchangeSent(exchange.getContext(), exchange,
endpoint, timeTaken);
+ }
+ }
+
+ /**
+ * Sends the exchange to the producer: fires the {@link
CamelEvent.ExchangeSendingEvent}, calls the producer with
+ * the {@link SendHook} around it, and fires the {@link
CamelEvent.ExchangeSentEvent} before the callback is called.
+ * <p/>
+ * An exception the producer throws, instead of calling the callback, is
thrown to the caller, without the sent
+ * event, as when the producer is called directly.
+ *
+ * @param exchange the exchange
+ * @param endpoint the endpoint of the producer
+ * @param producer the producer
+ * @param callback the callback, called exactly once when the send is done
+ * @return whether the send completed synchronously (the same
value the callback was called with)
+ */
+ public static boolean send(Exchange exchange, Endpoint endpoint,
AsyncProcessor producer, AsyncCallback callback) {
+ final StopWatch watch = notifySending(exchange, endpoint);
+ if (watch == null) {
+ return process(exchange, endpoint, producer, callback);
+ }
+ return process(exchange, endpoint, producer, doneSync -> {
+ try {
+ notifySent(exchange, endpoint, watch);
+ } finally {
+ callback.done(doneSync);
+ }
+ });
+ }
+
+ /**
+ * Calls the producer with the {@link SendHook} of the {@link
CamelContext} around it. Without a hook this is
+ * {@code producer.process(exchange, callback)}.
+ * <p/>
+ * With a hook, see {@link SendHook} for what the hook can do. An
exception the producer throws, instead of calling
+ * the callback, is set on the exchange, the hook's {@link
SendHook#afterSend(Exchange, Endpoint)} is called, and
+ * the exception is thrown to the caller, as without a hook.
+ *
+ * @param exchange the exchange
+ * @param endpoint the endpoint of the producer
+ * @param producer the producer
+ * @param callback the callback, called exactly once when the send is done
+ * @return whether the send completed synchronously (the same
value the callback was called with)
+ */
+ public static boolean process(Exchange exchange, Endpoint endpoint,
AsyncProcessor producer, AsyncCallback callback) {
+ final SendHook hook =
exchange.getContext().getCamelContextExtension().getSendHook();
+ if (hook == null) {
+ return producer.process(exchange, callback);
+ }
+ return new HookedSend(hook, exchange, endpoint, producer,
callback).send();
+ }
+
+ /**
+ * Returns the producer as it is when no {@link SendHook} is installed, or
else a producer that calls it with the
+ * hook around it (see {@link #newSendHookProducer(AsyncProducer)}).
+ *
+ * @param context the camel context
+ * @param producer the producer
+ * @return the producer, or a producer with the hook around it
+ */
+ public static AsyncProducer withSendHook(CamelContext context,
AsyncProducer producer) {
+ if (context.getCamelContextExtension().getSendHook() == null) {
+ return producer;
+ }
+ return newSendHookProducer(producer);
+ }
+
+ /**
+ * Creates a producer that calls the given producer with the {@link
SendHook} around it, as
+ * {@link #process(Exchange, Endpoint, AsyncProcessor, AsyncCallback)}
does. It looks up the hook on every send, so
+ * it can be kept (for example as the output of an error handler) when the
hook is removed later. It passes start,
+ * stop, shutdown and close on to the given producer.
+ *
+ * @param producer the producer
+ * @return a producer with the hook around it
+ */
+ public static AsyncProducer newSendHookProducer(AsyncProducer producer) {
+ return new SendHookProducer(producer);
+ }
+
+ /**
+ * One send with a hook. The hook is called before the producer, the
producer is called right away, after a delay
+ * (that this thread waits for), or not at all, and then the hook is
called after it and the callback is called,
+ * each exactly once.
+ */
+ private static final class HookedSend implements AsyncCallback {
+
+ private final SendHook hook;
+ private final Exchange exchange;
+ private final Endpoint endpoint;
+ private final AsyncProcessor producer;
+ private final AsyncCallback callback;
+
+ private HookedSend(SendHook hook, Exchange exchange, Endpoint
endpoint, AsyncProcessor producer,
+ AsyncCallback callback) {
+ this.hook = hook;
+ this.exchange = exchange;
+ this.endpoint = endpoint;
+ this.producer = producer;
+ this.callback = callback;
+ }
+
+ boolean send() {
+ // an exchange can be sent with an exception on it already, which
does not stop the send
+ final Exception existing = exchange.getException();
+ final long delay;
+ try {
+ delay = hook.beforeSend(exchange, endpoint);
+ } catch (Exception e) {
+ // the hook failed: the producer is not called, and there is
no after
+ exchange.setException(e);
+ callback.done(true);
+ return true;
+ }
+
+ final Exception exception = exchange.getException();
+ if (exception != null && exception != existing || delay ==
SendHook.SKIP) {
+ // the hook stopped (or skipped) the send, so the producer is
not called
+ return completeWithoutProducer();
+ }
+ if (delay < 0) {
+ // the other negative values are reserved, so a hook that
returns one fails the send
+ exchange.setException(new IllegalArgumentException(
+ "SendHook.beforeSend returned " + delay + ": negative
values other than SKIP are reserved"));
+ return completeWithoutProducer();
+ }
+
+ if (delay > 0) {
+ // wait on this thread, as for a slow endpoint (through
TimeSource, so a virtual clock can run it)
+ try {
+ TimeSource.sleep(delay);
+ } catch (InterruptedException e) {
+ // the send fails and the thread keeps its interrupted
status; the exception marks the exchange as
+ // interrupted, so Camel stops routing it, as when the
wait of the redelivery error handler is
+ // interrupted
+ Thread.currentThread().interrupt();
+ exchange.setException(e);
+ return completeWithoutProducer();
+ } catch (RuntimeException e) {
+ // the clock failed: as for a producer that throws
+ exchange.setException(e);
+ afterSend();
+ throw e;
+ }
+ }
+
+ try {
+ return producer.process(exchange, this);
+ } catch (RuntimeException e) {
+ // the caller handles a producer that throws as it does
without a hook
+ exchange.setException(e);
+ afterSend();
+ throw e;
+ }
+ }
+
+ private boolean completeWithoutProducer() {
+ afterSend();
+ callback.done(true);
+ return true;
+ }
+
+ /**
+ * The producer is done.
+ */
+ @Override
+ public void done(boolean doneSync) {
+ afterSend();
+ callback.done(doneSync);
+ }
+
+ private void afterSend() {
+ try {
+ hook.afterSend(exchange, endpoint);
+ } catch (Exception e) {
+ Exception existing = exchange.getException();
+ if (existing == null) {
+ exchange.setException(e);
+ } else if (existing != e) {
+ existing.addSuppressed(e);
+ }
+ }
+ }
+ }
+
+ /**
+ * A producer that calls another producer with the {@link SendHook} around
it. It passes its lifecycle on to that
+ * producer, so as the output of an error handler it is shut down with the
error handler as the producer would be.
+ * (It is not {@link org.apache.camel.SuspendableService}: an error
handler does not suspend its output.)
+ */
+ private static final class SendHookProducer implements AsyncProducer,
ShutdownableService {
+
+ private final AsyncProducer producer;
+
+ private SendHookProducer(AsyncProducer producer) {
+ this.producer = producer;
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ return SendHelper.process(exchange, producer.getEndpoint(),
producer, callback);
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+
PluginHelper.getAsyncProcessorAwaitManager(exchange.getContext()).process(this,
exchange);
+ }
+
+ @Override
+ public CompletableFuture<Exchange> processAsync(Exchange exchange) {
+ AsyncCallbackToCompletableFutureAdapter<Exchange> callback
+ = new AsyncCallbackToCompletableFutureAdapter<>(exchange);
+ process(exchange, callback);
+ return callback.getFuture();
+ }
+
+ @Override
+ public Endpoint getEndpoint() {
+ return producer.getEndpoint();
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return producer.isSingleton();
+ }
+
+ @Override
+ public void build() {
+ producer.build();
+ }
+
+ @Override
+ public void init() {
+ producer.init();
+ }
+
+ @Override
+ public void start() {
+ producer.start();
+ }
+
+ @Override
+ public void stop() {
+ producer.stop();
+ }
+
+ @Override
+ public void shutdown() {
+ // as for the producer itself: stopped, and shut down if it can be
+ ServiceHelper.stopAndShutdownService(producer);
+ }
+
+ @Override
+ public void close() throws IOException {
+ producer.close();
+ }
+
+ @Override
+ public String toString() {
+ return producer.toString();
+ }
+ }
+}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/cache/DefaultProducerCache.java
b/core/camel-support/src/main/java/org/apache/camel/support/cache/DefaultProducerCache.java
index 912afadeecac..4eaec5a6bd46 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/cache/DefaultProducerCache.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/cache/DefaultProducerCache.java
@@ -36,8 +36,8 @@ import org.apache.camel.spi.ProducerCache;
import org.apache.camel.spi.SharedInternalProcessor;
import org.apache.camel.support.CamelContextHelper;
import org.apache.camel.support.DefaultEndpointUtilizationStatistics;
-import org.apache.camel.support.EventHelper;
import org.apache.camel.support.PluginHelper;
+import org.apache.camel.support.SendHelper;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.util.StopWatch;
@@ -170,24 +170,18 @@ public class DefaultProducerCache extends ServiceSupport
implements ProducerCach
StopWatch watch = null;
try {
if (eventNotifierEnabled &&
camelContext.getCamelContextExtension().isEventNotificationApplicable()) {
- boolean sending =
EventHelper.notifyExchangeSending(exchange.getContext(), exchange, endpoint);
- if (sending) {
- watch = new StopWatch();
- }
+ watch = SendHelper.notifySending(exchange, endpoint);
}
- // invoke the synchronous method
- sharedInternalProcessor.process(exchange, producer,
resultProcessor);
+ // invoke the synchronous method (with the send hook around
the producer if any)
+ sharedInternalProcessor.process(exchange,
SendHelper.withSendHook(camelContext, producer), resultProcessor);
} catch (Exception e) {
// ensure exceptions is caught and set on the exchange
exchange.setException(e);
} finally {
// emit event that the exchange was sent to the endpoint
- if (watch != null) {
- long timeTaken = watch.taken();
- EventHelper.notifyExchangeSent(exchange.getContext(),
exchange, endpoint, timeTaken);
- }
+ SendHelper.notifySent(exchange, endpoint, watch);
}
return exchange;
} finally {
@@ -266,27 +260,19 @@ public class DefaultProducerCache extends ServiceSupport
implements ProducerCach
}
try {
- // record timing for sending the exchange using the producer
- StopWatch watch;
+ // emit event that the exchange is being sent, and record timing
for sending the exchange using the producer
+ final StopWatch watch;
if (eventNotifierEnabled &&
camelContext.getCamelContextExtension().isEventNotificationApplicable()) {
- boolean sending =
EventHelper.notifyExchangeSending(exchange.getContext(), exchange, endpoint);
- if (sending) {
- watch = new StopWatch();
- } else {
- watch = null;
- }
+ watch = SendHelper.notifySending(exchange, endpoint);
} else {
watch = null;
}
- // invoke the callback
- return producerCallback.doInAsyncProducer(producer, exchange,
doneSync -> {
+ // invoke the callback, which calls the producer (with the send
hook around it if any)
+ return
producerCallback.doInAsyncProducer(SendHelper.withSendHook(camelContext,
producer), exchange, doneSync -> {
try {
- if (watch != null) {
- long timeTaken = watch.taken();
- // emit event that the exchange was sent to the
endpoint
- EventHelper.notifyExchangeSent(exchange.getContext(),
exchange, endpoint, timeTaken);
- }
+ // emit event that the exchange was sent to the endpoint
+ SendHelper.notifySent(exchange, endpoint, watch);
// release back to the pool
releaseProducer(endpoint, producer);
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/cache/EventNotifierCallback.java
b/core/camel-support/src/main/java/org/apache/camel/support/cache/EventNotifierCallback.java
index 4d6f9eb312b2..38cb09ae0494 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/cache/EventNotifierCallback.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/cache/EventNotifierCallback.java
@@ -19,7 +19,7 @@ package org.apache.camel.support.cache;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
-import org.apache.camel.support.EventHelper;
+import org.apache.camel.support.SendHelper;
import org.apache.camel.util.StopWatch;
/**
@@ -36,20 +36,12 @@ class EventNotifierCallback implements AsyncCallback {
this.originalCallback = originalCallback;
this.exchange = exchange;
this.endpoint = endpoint;
- final boolean sending =
EventHelper.notifyExchangeSending(exchange.getContext(), exchange, endpoint);
- if (sending) {
- this.watch = new StopWatch();
- } else {
- this.watch = null;
- }
+ this.watch = SendHelper.notifySending(exchange, endpoint);
}
@Override
public void done(boolean doneSync) {
- if (watch != null) {
- long timeTaken = watch.taken();
- EventHelper.notifyExchangeSent(exchange.getContext(), exchange,
endpoint, timeTaken);
- }
+ SendHelper.notifySent(exchange, endpoint, watch);
originalCallback.done(doneSync);
}
}
