You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2015/07/09 10:30:54 UTC
camel git commit: CAMEL-8944: Routing engine - When UoW done let the
route context be present so MDC logging is possible
Repository: camel
Updated Branches:
refs/heads/master abec14bf6 -> af0486a0b
CAMEL-8944: Routing engine - When UoW done let the route context be present so MDC logging is possible
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/af0486a0
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/af0486a0
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/af0486a0
Branch: refs/heads/master
Commit: af0486a0beecd4876357e987836fc989b2ed155a
Parents: abec14b
Author: Claus Ibsen <da...@apache.org>
Authored: Thu Jul 9 10:26:47 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Thu Jul 9 10:26:47 2015 +0200
----------------------------------------------------------------------
.../apache/camel/impl/DefaultRouteContext.java | 6 +-
.../apache/camel/model/AggregateDefinition.java | 6 +-
.../camel/model/OnCompletionDefinition.java | 3 +-
.../camel/model/ResequenceDefinition.java | 9 +-
.../apache/camel/model/WireTapDefinition.java | 3 +-
.../camel/processor/CamelInternalProcessor.java | 43 ++++++---
.../camel/processor/MulticastProcessor.java | 9 +-
.../MDCOnCompletionOnCompletionTest.java | 94 ++++++++++++++++++++
8 files changed, 135 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/af0486a0/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java
index 52e4328..089bdea 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java
@@ -156,14 +156,12 @@ public class DefaultRouteContext implements RouteContext {
if (!eventDrivenProcessors.isEmpty()) {
Processor target = Pipeline.newInstance(getCamelContext(), eventDrivenProcessors);
+ // force creating the route id so its known ahead of the route is started
String routeId = route.idOrCreate(getCamelContext().getNodeIdFactory());
// and wrap it in a unit of work so the UoW is on the top, so the entire route will be in the same UoW
CamelInternalProcessor internal = new CamelInternalProcessor(target);
- internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeId));
-
- // and then in route context so we can keep track which route this is at runtime
- internal.addAdvice(new CamelInternalProcessor.RouteContextAdvice(this));
+ internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(this));
// and then optionally add route policy processor if a custom policy is set
List<RoutePolicy> routePolicyList = getRoutePolicyList();
http://git-wip-us.apache.org/repos/asf/camel/blob/af0486a0/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java b/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
index cfcb027..12bebb4 100644
--- a/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/AggregateDefinition.java
@@ -41,7 +41,6 @@ import org.apache.camel.processor.aggregate.AggregationStrategy;
import org.apache.camel.processor.aggregate.AggregationStrategyBeanAdapter;
import org.apache.camel.processor.aggregate.GroupedExchangeAggregationStrategy;
import org.apache.camel.processor.aggregate.OptimisticLockRetryPolicy;
-import org.apache.camel.processor.aggregate.PreCompletionAwareAggregationStrategy;
import org.apache.camel.spi.AggregationRepository;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.RouteContext;
@@ -168,12 +167,9 @@ public class AggregateDefinition extends ProcessorDefinition<AggregateDefinition
protected AggregateProcessor createAggregator(RouteContext routeContext) throws Exception {
Processor childProcessor = this.createChildProcessor(routeContext, true);
- String routeId = routeContext.getRoute().idOrCreate(routeContext.getCamelContext().getNodeIdFactory());
-
// wrap the aggregate route in a unit of work processor
CamelInternalProcessor internal = new CamelInternalProcessor(childProcessor);
- internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeId));
- internal.addAdvice(new CamelInternalProcessor.RouteContextAdvice(routeContext));
+ internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeContext));
Expression correlation = getExpression().createExpression(routeContext);
AggregationStrategy strategy = createAggregationStrategy(routeContext);
http://git-wip-us.apache.org/repos/asf/camel/blob/af0486a0/camel-core/src/main/java/org/apache/camel/model/OnCompletionDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/OnCompletionDefinition.java b/camel-core/src/main/java/org/apache/camel/model/OnCompletionDefinition.java
index de76971..c44d0d0 100644
--- a/camel-core/src/main/java/org/apache/camel/model/OnCompletionDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/OnCompletionDefinition.java
@@ -133,8 +133,7 @@ public class OnCompletionDefinition extends ProcessorDefinition<OnCompletionDefi
// wrap the on completion route in a unit of work processor
CamelInternalProcessor internal = new CamelInternalProcessor(childProcessor);
- internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeId));
- internal.addAdvice(new CamelInternalProcessor.RouteContextAdvice(routeContext));
+ internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeContext));
onCompletions.put(routeId, internal);
http://git-wip-us.apache.org/repos/asf/camel/blob/af0486a0/camel-core/src/main/java/org/apache/camel/model/ResequenceDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/ResequenceDefinition.java b/camel-core/src/main/java/org/apache/camel/model/ResequenceDefinition.java
index 9e205f4..c6b6ed5 100644
--- a/camel-core/src/main/java/org/apache/camel/model/ResequenceDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/ResequenceDefinition.java
@@ -352,10 +352,8 @@ public class ResequenceDefinition extends ProcessorDefinition<ResequenceDefiniti
Expression expression = getExpression().createExpression(routeContext);
// and wrap in unit of work
- String routeId = routeContext.getRoute().idOrCreate(routeContext.getCamelContext().getNodeIdFactory());
CamelInternalProcessor internal = new CamelInternalProcessor(processor);
- internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeId));
- internal.addAdvice(new CamelInternalProcessor.RouteContextAdvice(routeContext));
+ internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeContext));
ObjectHelper.notNull(config, "config", this);
ObjectHelper.notNull(expression, "expression", this);
@@ -385,11 +383,8 @@ public class ResequenceDefinition extends ProcessorDefinition<ResequenceDefiniti
Processor processor = this.createChildProcessor(routeContext, true);
Expression expression = getExpression().createExpression(routeContext);
- // and wrap in unit of work
- String routeId = routeContext.getRoute().idOrCreate(routeContext.getCamelContext().getNodeIdFactory());
CamelInternalProcessor internal = new CamelInternalProcessor(processor);
- internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeId));
- internal.addAdvice(new CamelInternalProcessor.RouteContextAdvice(routeContext));
+ internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeContext));
ObjectHelper.notNull(config, "config", this);
ObjectHelper.notNull(expression, "expression", this);
http://git-wip-us.apache.org/repos/asf/camel/blob/af0486a0/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java b/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
index 3f53451..49631bb 100644
--- a/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/WireTapDefinition.java
@@ -108,9 +108,8 @@ public class WireTapDefinition<Type extends ProcessorDefinition<Type>> extends N
Processor target = wrapInErrorHandler(routeContext, producer);
// and wrap in unit of work
- String routeId = routeContext.getRoute().idOrCreate(routeContext.getCamelContext().getNodeIdFactory());
CamelInternalProcessor internal = new CamelInternalProcessor(target);
- internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeId));
+ internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeContext));
// is true bt default
boolean isCopy = getCopy() == null || getCopy();
http://git-wip-us.apache.org/repos/asf/camel/blob/af0486a0/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
index f8c71f4..856a225 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
@@ -388,7 +388,10 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor {
/**
* Advice to inject the current {@link RouteContext} into the {@link UnitOfWork} on the {@link Exchange}
+ *
+ * @deprecated this logic has been merged into {@link org.apache.camel.processor.CamelInternalProcessor.UnitOfWorkProcessorAdvice}
*/
+ @Deprecated
public static class RouteContextAdvice implements CamelInternalProcessorAdvice<UnitOfWork> {
private final RouteContext routeContext;
@@ -604,38 +607,56 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor {
*/
public static class UnitOfWorkProcessorAdvice implements CamelInternalProcessorAdvice<UnitOfWork> {
- private final String routeId;
+ private final RouteContext routeContext;
- public UnitOfWorkProcessorAdvice(String routeId) {
- this.routeId = routeId;
+ public UnitOfWorkProcessorAdvice(RouteContext routeContext) {
+ this.routeContext = routeContext;
}
@Override
public UnitOfWork before(Exchange exchange) throws Exception {
// if the exchange doesn't have from route id set, then set it if it originated
// from this unit of work
- if (routeId != null && exchange.getFromRouteId() == null) {
+ if (routeContext != null && exchange.getFromRouteId() == null) {
+ String routeId = routeContext.getRoute().idOrCreate(routeContext.getCamelContext().getNodeIdFactory());
exchange.setFromRouteId(routeId);
}
+ // only return UnitOfWork if we created a new as then its us that handle the lifecycle to done the created UoW
+ UnitOfWork created = null;
+
if (exchange.getUnitOfWork() == null) {
// If there is no existing UoW, then we should start one and
// terminate it once processing is completed for the exchange.
- UnitOfWork uow = createUnitOfWork(exchange);
- exchange.setUnitOfWork(uow);
- uow.start();
- return uow;
+ created = createUnitOfWork(exchange);
+ exchange.setUnitOfWork(created);
+ created.start();
}
- return null;
+ // for any exchange we should push/pop route context so we can keep track of which route we are routing
+ if (routeContext != null) {
+ UnitOfWork existing = exchange.getUnitOfWork();
+ if (existing != null) {
+ existing.pushRouteContext(routeContext);
+ }
+ }
+
+ return created;
}
@Override
public void after(Exchange exchange, UnitOfWork uow) throws Exception {
+ UnitOfWork existing = exchange.getUnitOfWork();
+
// execute done on uow if we created it, and the consumer is not doing it
if (uow != null) {
UnitOfWorkHelper.doneUow(uow, exchange);
}
+
+ // after UoW is done lets pop the route context which must be done on every existing UoW
+ if (routeContext != null && existing != null) {
+ existing.popRouteContext();
+ }
}
protected UnitOfWork createUnitOfWork(Exchange exchange) {
@@ -651,8 +672,8 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor {
private final UnitOfWork parent;
- public ChildUnitOfWorkProcessorAdvice(String routeId, UnitOfWork parent) {
- super(routeId);
+ public ChildUnitOfWorkProcessorAdvice(RouteContext routeContext, UnitOfWork parent) {
+ super(routeContext);
this.parent = parent;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/af0486a0/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
index e4a2ef8..0370305 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
@@ -1084,21 +1084,16 @@ public class MulticastProcessor extends ServiceSupport implements AsyncProcessor
* @return the unit of work processor
*/
protected Processor createUnitOfWorkProcessor(RouteContext routeContext, Processor processor, Exchange exchange) {
- String routeId = routeContext != null ? routeContext.getRoute().idOrCreate(routeContext.getCamelContext().getNodeIdFactory()) : null;
CamelInternalProcessor internal = new CamelInternalProcessor(processor);
// and wrap it in a unit of work so the UoW is on the top, so the entire route will be in the same UoW
UnitOfWork parent = exchange.getProperty(Exchange.PARENT_UNIT_OF_WORK, UnitOfWork.class);
if (parent != null) {
- internal.addAdvice(new CamelInternalProcessor.ChildUnitOfWorkProcessorAdvice(routeId, parent));
+ internal.addAdvice(new CamelInternalProcessor.ChildUnitOfWorkProcessorAdvice(routeContext, parent));
} else {
- internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeId));
+ internal.addAdvice(new CamelInternalProcessor.UnitOfWorkProcessorAdvice(routeContext));
}
- // and then in route context so we can keep track which route this is at runtime
- if (routeContext != null) {
- internal.addAdvice(new CamelInternalProcessor.RouteContextAdvice(routeContext));
- }
return internal;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/af0486a0/camel-core/src/test/java/org/apache/camel/processor/MDCOnCompletionOnCompletionTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/MDCOnCompletionOnCompletionTest.java b/camel-core/src/test/java/org/apache/camel/processor/MDCOnCompletionOnCompletionTest.java
new file mode 100644
index 0000000..d46f6f9
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/MDCOnCompletionOnCompletionTest.java
@@ -0,0 +1,94 @@
+/**
+ * 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 org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.support.SynchronizationAdapter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
+/**
+ * @version
+ */
+public class MDCOnCompletionOnCompletionTest extends ContextTestSupport {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MDCOnCompletionOnCompletionTest.class);
+
+ public void testMDC() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMinimumMessageCount(1);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ // enable MDC
+ context.setUseMDCLogging(true);
+
+ from("timer:foo?period=5000").routeId("route-a")
+ .setBody().constant("Hello World")
+ .onCompletion()
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ exchange.addOnCompletion(new MyOnCompletion());
+ }
+ })
+ .end()
+ .to("log:foo")
+ .to("direct:b");
+
+ from("direct:b").routeId("route-b")
+ .process(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ assertEquals("route-b", MDC.get("camel.routeId"));
+ assertEquals(exchange.getExchangeId(), MDC.get("camel.exchangeId"));
+ assertEquals(exchange.getIn().getMessageId(), MDC.get("camel.messageId"));
+
+ MDC.put("custom.id", "1");
+ LOG.info("From processor in route-b");
+ }
+ })
+ .to("log:bar")
+ .to("mock:result");
+ }
+ };
+ }
+
+ private class MyOnCompletion extends SynchronizationAdapter {
+
+ @Override
+ public void onDone(Exchange exchange) {
+ assertEquals("route-a", MDC.get("camel.routeId"));
+ assertEquals(exchange.getExchangeId(), MDC.get("camel.exchangeId"));
+ assertEquals(exchange.getIn().getMessageId(), MDC.get("camel.messageId"));
+
+ assertEquals("1", MDC.get("custom.id"));
+
+ LOG.info("From onCompletion after route-a");
+ }
+ }
+}