You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2023/05/15 08:02:44 UTC
[camel] branch main updated: CAMEL-19060: use an extension for the Synchronization
This is an automated email from the ASF dual-hosted git repository.
orpiske 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 6c4b58f0d58 CAMEL-19060: use an extension for the Synchronization
6c4b58f0d58 is described below
commit 6c4b58f0d58df4e0bf5349c0249ed86d2fac17b5
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Fri May 5 10:55:02 2023 +0200
CAMEL-19060: use an extension for the Synchronization
This helps the code a bit when under heavy contention.
---
.../java/org/apache/camel/spi/Synchronization.java | 10 ++
.../camel/spi/SynchronizationRouteAware.java | 8 +-
.../camel/impl/engine/CamelInternalProcessor.java | 19 +++-
.../camel/processor/OnCompletionProcessor.java | 105 ++++++++++++---------
.../processor/RouteAwareSynchronizationTest.java | 29 +++---
.../camel/support/SynchronizationAdapter.java | 15 +--
.../org/apache/camel/support/UnitOfWorkHelper.java | 34 ++++---
7 files changed, 130 insertions(+), 90 deletions(-)
diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/Synchronization.java b/core/camel-api/src/main/java/org/apache/camel/spi/Synchronization.java
index 1bccd865f90..ced9c7232a0 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/Synchronization.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/Synchronization.java
@@ -40,4 +40,14 @@ public interface Synchronization {
* @param exchange the exchange being processed
*/
void onFailure(Exchange exchange);
+
+ /**
+ * Get an optional {@link SynchronizationRouteAware} for this synchronization
+ *
+ * @return An instance of {@link SynchronizationRouteAware} or null if unset for this synchronization
+ */
+ default SynchronizationRouteAware getRouteSynchronization() {
+ return null;
+ }
+
}
diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/SynchronizationRouteAware.java b/core/camel-api/src/main/java/org/apache/camel/spi/SynchronizationRouteAware.java
index 74448a0ae21..28f834f1cd3 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/SynchronizationRouteAware.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/SynchronizationRouteAware.java
@@ -20,9 +20,9 @@ import org.apache.camel.Exchange;
import org.apache.camel.Route;
/**
- * An extended {@link org.apache.camel.spi.Synchronization} which is route aware.
+ * An extension to {@link org.apache.camel.spi.Synchronization} which provides route awareness capabilities.
*/
-public interface SynchronizationRouteAware extends Synchronization {
+public interface SynchronizationRouteAware {
/**
* Invoked before the {@link org.apache.camel.Exchange} is being routed by the given route.
@@ -45,8 +45,8 @@ public interface SynchronizationRouteAware extends Synchronization {
* This invocation happens before these callbacks:
* <ul>
* <li>The consumer of the route writes any response back to the caller (if in InOut mode)</li>
- * <li>The UoW is done calling either {@link #onComplete(org.apache.camel.Exchange)} or
- * {@link #onFailure(org.apache.camel.Exchange)}</li>
+ * <li>The UoW is done calling either {@link Synchronization#onComplete(org.apache.camel.Exchange)} or
+ * {@link Synchronization#onFailure(org.apache.camel.Exchange)}</li>
* </ul>
* This allows custom logic to be executed after all routing is done, but before the
* {@link org.apache.camel.Consumer} prepares and writes any data back to the caller (if in InOut mode).
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
index b6622a4368b..1dac8ebc18f 100644
--- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
@@ -51,6 +51,7 @@ import org.apache.camel.spi.RoutePolicy;
import org.apache.camel.spi.ShutdownStrategy;
import org.apache.camel.spi.StreamCachingStrategy;
import org.apache.camel.spi.Synchronization;
+import org.apache.camel.spi.SynchronizationRouteAware;
import org.apache.camel.spi.Tracer;
import org.apache.camel.spi.Transformer;
import org.apache.camel.spi.UnitOfWork;
@@ -1101,10 +1102,20 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor implements In
}
@Override
- public void onAfterRoute(Route route, Exchange exchange) {
- if (routeId.equals(route.getId())) {
- tracer.traceAfterRoute(node, exchange);
- }
+ public SynchronizationRouteAware getRouteSynchronization() {
+ return new SynchronizationRouteAware() {
+ @Override
+ public void onBeforeRoute(Route route, Exchange exchange) {
+ // NO-OP
+ }
+
+ @Override
+ public void onAfterRoute(Route route, Exchange exchange) {
+ if (routeId.equals(route.getId())) {
+ tracer.traceAfterRoute(node, exchange);
+ }
+ }
+ };
}
@Override
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
index fbd69fd9dc5..0c3abe1f9cf 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
@@ -34,6 +34,7 @@ import org.apache.camel.Route;
import org.apache.camel.Traceable;
import org.apache.camel.spi.IdAware;
import org.apache.camel.spi.RouteIdAware;
+import org.apache.camel.spi.SynchronizationRouteAware;
import org.apache.camel.support.AsyncProcessorSupport;
import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.SynchronizationAdapter;
@@ -261,17 +262,27 @@ public class OnCompletionProcessor extends AsyncProcessorSupport implements Trac
}
@Override
- @SuppressWarnings("unchecked")
- public void onAfterRoute(Route route, Exchange exchange) {
- // route scope = remember we have been at this route
- if (routeScoped && route.getRouteId().equals(routeId)) {
- List<String> routeIds = exchange.getProperty(ExchangePropertyKey.ON_COMPLETION_ROUTE_IDS, List.class);
- if (routeIds == null) {
- routeIds = new ArrayList<>();
- exchange.setProperty(ExchangePropertyKey.ON_COMPLETION_ROUTE_IDS, routeIds);
+ public SynchronizationRouteAware getRouteSynchronization() {
+ return new SynchronizationRouteAware() {
+ @Override
+ public void onBeforeRoute(Route route, Exchange exchange) {
+ // NO-OP
}
- routeIds.add(route.getRouteId());
- }
+
+ @Override
+ public void onAfterRoute(Route route, Exchange exchange) {
+ // route scope = remember we have been at this route
+ if (routeScoped && route.getRouteId().equals(routeId)) {
+ @SuppressWarnings("unchecked")
+ List<String> routeIds = exchange.getProperty(ExchangePropertyKey.ON_COMPLETION_ROUTE_IDS, List.class);
+ if (routeIds == null) {
+ routeIds = new ArrayList<>();
+ exchange.setProperty(ExchangePropertyKey.ON_COMPLETION_ROUTE_IDS, routeIds);
+ }
+ routeIds.add(route.getRouteId());
+ }
+ }
+ };
}
@Override
@@ -408,47 +419,57 @@ public class OnCompletionProcessor extends AsyncProcessorSupport implements Trac
}
@Override
- public void onAfterRoute(Route route, Exchange exchange) {
- LOG.debug("onAfterRoute from Route {}", route.getRouteId());
- // route scope = should be from this route
- if (routeScoped && !route.getRouteId().equals(routeId)) {
- return;
- }
+ public SynchronizationRouteAware getRouteSynchronization() {
+ return new SynchronizationRouteAware() {
+ @Override
+ public void onBeforeRoute(Route route, Exchange exchange) {
+ // NO-OP
+ }
- // global scope = should be from the original route
- if (!routeScoped && (!route.getRouteId().equals(routeId) || !exchange.getFromRouteId().equals(routeId))) {
- return;
- }
+ @Override
+ public void onAfterRoute(Route route, Exchange exchange) {
+ LOG.debug("onAfterRoute from Route {}", route.getRouteId());
+ // route scope = should be from this route
+ if (routeScoped && !route.getRouteId().equals(routeId)) {
+ return;
+ }
- if (exchange.isFailed() && onCompleteOnly) {
- return;
- }
+ // global scope = should be from the original route
+ if (!routeScoped && (!route.getRouteId().equals(routeId) || !exchange.getFromRouteId().equals(routeId))) {
+ return;
+ }
- if (!exchange.isFailed() && onFailureOnly) {
- return;
- }
+ if (exchange.isFailed() && onCompleteOnly) {
+ return;
+ }
- if (onWhen != null && !onWhen.matches(exchange)) {
- // predicate did not match so do not route the onComplete
- return;
- }
+ if (!exchange.isFailed() && onFailureOnly) {
+ return;
+ }
- // must use a copy as we don't want it to cause side effects of the original exchange
- final Exchange copy = prepareExchange(exchange);
+ if (onWhen != null && !onWhen.matches(exchange)) {
+ // predicate did not match so do not route the onComplete
+ return;
+ }
- if (executorService != null) {
- executorService.submit(new Callable<Exchange>() {
- public Exchange call() throws Exception {
+ // must use a copy as we don't want it to cause side effects of the original exchange
+ final Exchange copy = prepareExchange(exchange);
+
+ if (executorService != null) {
+ executorService.submit(new Callable<Exchange>() {
+ public Exchange call() throws Exception {
+ LOG.debug("Processing onAfterRoute: {}", copy);
+ doProcess(processor, copy);
+ return copy;
+ }
+ });
+ } else {
+ // run without thread-pool
LOG.debug("Processing onAfterRoute: {}", copy);
doProcess(processor, copy);
- return copy;
}
- });
- } else {
- // run without thread-pool
- LOG.debug("Processing onAfterRoute: {}", copy);
- doProcess(processor, copy);
- }
+ }
+ };
}
@Override
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/RouteAwareSynchronizationTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/RouteAwareSynchronizationTest.java
index cf76f57c53c..03b6dec39a7 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/RouteAwareSynchronizationTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/RouteAwareSynchronizationTest.java
@@ -24,6 +24,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.Route;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.Synchronization;
import org.apache.camel.spi.SynchronizationRouteAware;
import org.junit.jupiter.api.Test;
@@ -72,18 +73,7 @@ public class RouteAwareSynchronizationTest extends ContextTestSupport {
};
}
- private static final class MyRouteAware implements SynchronizationRouteAware {
-
- @Override
- public void onBeforeRoute(Route route, Exchange exchange) {
- EVENTS.add("onBeforeRoute-" + route.getId());
- }
-
- @Override
- public void onAfterRoute(Route route, Exchange exchange) {
- EVENTS.add("onAfterRoute-" + route.getId());
- }
-
+ private static final class MyRouteAware implements Synchronization {
@Override
public void onComplete(Exchange exchange) {
EVENTS.add("onComplete");
@@ -93,5 +83,20 @@ public class RouteAwareSynchronizationTest extends ContextTestSupport {
public void onFailure(Exchange exchange) {
EVENTS.add("onFailure");
}
+
+ @Override
+ public SynchronizationRouteAware getRouteSynchronization() {
+ return new SynchronizationRouteAware() {
+ @Override
+ public void onBeforeRoute(Route route, Exchange exchange) {
+ EVENTS.add("onBeforeRoute-" + route.getId());
+ }
+
+ @Override
+ public void onAfterRoute(Route route, Exchange exchange) {
+ EVENTS.add("onAfterRoute-" + route.getId());
+ }
+ };
+ }
}
}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/SynchronizationAdapter.java b/core/camel-support/src/main/java/org/apache/camel/support/SynchronizationAdapter.java
index 7d9ce2c67f0..c219095444c 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/SynchronizationAdapter.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/SynchronizationAdapter.java
@@ -18,16 +18,13 @@ package org.apache.camel.support;
import org.apache.camel.Exchange;
import org.apache.camel.Ordered;
-import org.apache.camel.Route;
import org.apache.camel.spi.Synchronization;
-import org.apache.camel.spi.SynchronizationRouteAware;
import org.apache.camel.spi.SynchronizationVetoable;
/**
* Simple {@link Synchronization} adapter with empty methods for easier overriding of single methods.
*/
-public class SynchronizationAdapter implements SynchronizationVetoable, Ordered, SynchronizationRouteAware {
-
+public class SynchronizationAdapter implements SynchronizationVetoable, Ordered, Synchronization {
@Override
public void onComplete(Exchange exchange) {
onDone(exchange);
@@ -54,16 +51,6 @@ public class SynchronizationAdapter implements SynchronizationVetoable, Ordered,
return 0;
}
- @Override
- public void onBeforeRoute(Route route, Exchange exchange) {
- // noop
- }
-
- @Override
- public void onAfterRoute(Route route, Exchange exchange) {
- // noop
- }
-
@Override
public void beforeHandover(Exchange target) {
// noop
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 38bca852d1a..69c1be2ca3a 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
@@ -103,18 +103,21 @@ public final class UnitOfWorkHelper {
// work on a copy of the list to avoid any modification which may cause ConcurrentModificationException
List<Synchronization> copy = new ArrayList<>(synchronizations);
- // reverse so we invoke it FILO style instead of FIFO
- Collections.reverse(copy);
- // and honor if any was ordered by sorting it accordingly
- copy.sort(OrderedComparator.get());
+ if (copy.size() > 1) {
+ // reverse so we invoke it FILO style instead of FIFO
+ Collections.reverse(copy);
+ // and honor if any was ordered by sorting it accordingly
+ copy.sort(OrderedComparator.get());
+ }
// invoke synchronization callbacks
for (Synchronization synchronization : copy) {
- if (synchronization instanceof SynchronizationRouteAware) {
+ final SynchronizationRouteAware routeSynchronization = synchronization.getRouteSynchronization();
+ if (routeSynchronization != null) {
try {
log.trace("Invoking synchronization.onBeforeRoute: {} with {}", synchronization, exchange);
- ((SynchronizationRouteAware) synchronization).onBeforeRoute(route, exchange);
- } catch (Throwable e) {
+ routeSynchronization.onBeforeRoute(route, exchange);
+ } catch (Exception e) {
// must catch exceptions to ensure all synchronizations have a chance to run
log.warn("Exception occurred during onBeforeRoute. This exception will be ignored.", e);
}
@@ -127,18 +130,21 @@ public final class UnitOfWorkHelper {
// work on a copy of the list to avoid any modification which may cause ConcurrentModificationException
List<Synchronization> copy = new ArrayList<>(synchronizations);
- // reverse so we invoke it FILO style instead of FIFO
- Collections.reverse(copy);
- // and honor if any was ordered by sorting it accordingly
- copy.sort(OrderedComparator.get());
+ if (copy.size() > 1) {
+ // reverse so we invoke it FILO style instead of FIFO
+ Collections.reverse(copy);
+ // and honor if any was ordered by sorting it accordingly
+ copy.sort(OrderedComparator.get());
+ }
// invoke synchronization callbacks
for (Synchronization synchronization : copy) {
- if (synchronization instanceof SynchronizationRouteAware) {
+ final SynchronizationRouteAware routeSynchronization = synchronization.getRouteSynchronization();
+ if (routeSynchronization != null) {
try {
log.trace("Invoking synchronization.onAfterRoute: {} with {}", synchronization, exchange);
- ((SynchronizationRouteAware) synchronization).onAfterRoute(route, exchange);
- } catch (Throwable e) {
+ routeSynchronization.onAfterRoute(route, exchange);
+ } catch (Exception e) {
// must catch exceptions to ensure all synchronizations have a chance to run
log.warn("Exception occurred during onAfterRoute. This exception will be ignored.", e);
}