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 2009/11/14 18:47:19 UTC
svn commit: r836224 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/ main/java/org/apache/camel/impl/
main/java/org/apache/camel/model/ main/java/org/apache/camel/processor/
main/java/org/apache/camel/processor/interceptor/ main/java/org/a...
Author: davsclaus
Date: Sat Nov 14 17:47:18 2009
New Revision: 836224
URL: http://svn.apache.org/viewvc?rev=836224&view=rev
Log:
CAMEL-2051: Channel now have a childDefinition for more fine grained which is needed by tracer. Fixed tracer to handle child routes from splitter/aggregator etc.
Added:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultTracedRouteNodes.java (with props)
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnCompletionRouteNode.java
- copied, changed from r836130, camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteNode.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnExceptionRouteNode.java (with props)
camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TracedRouteNodes.java (contents, props changed)
- copied, changed from r836130, camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TraceableUnitOfWork.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/TraceDelayerTest.java
- copied, changed from r836130, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DelayerTest.java
Removed:
camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TraceableUnitOfWork.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/Channel.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteNode.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultUnitOfWork.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DefaultChannel.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultTraceEventMessage.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultTraceFormatter.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/TraceInterceptor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/spi/RouteContext.java
camel/trunk/camel-core/src/main/java/org/apache/camel/spi/UnitOfWork.java
camel/trunk/camel-core/src/test/java/org/apache/camel/impl/CustomIdFactoryTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DefaultErrorHandlerOnExceptionTraceTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/TraceableUnitOfWorkTest.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/Channel.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/Channel.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/Channel.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/Channel.java Sat Nov 14 17:47:18 2009
@@ -78,13 +78,21 @@
/**
* Initializes the channel.
*
- * @param outputDefinition the route defintion the {@link Channel} represents
+ * @param outputDefinition the route definition the {@link Channel} represents
* @param routeContext the route context
- * @throws Exception is thrown if some error occured
+ * @throws Exception is thrown if some error occurred
*/
void initChannel(ProcessorDefinition<?> outputDefinition, RouteContext routeContext) throws Exception;
/**
+ * If the initialized output definition contained outputs (children) then we need to
+ * set the child so we can leverage fine grained tracing
+ *
+ * @param child the child
+ */
+ void setChildDefinition(ProcessorDefinition<?> child);
+
+ /**
* Gets the wrapped output that at runtime should be delegated to.
*
* @return the output to route the {@link Exchange} to
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java Sat Nov 14 17:47:18 2009
@@ -18,8 +18,10 @@
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
@@ -44,6 +46,7 @@
* @version $Revision$
*/
public class DefaultRouteContext implements RouteContext {
+ private final Map<ProcessorDefinition<?>, AtomicInteger> nodeIndex = new HashMap<ProcessorDefinition<?>, AtomicInteger>();
private final RouteDefinition route;
private FromDefinition from;
private final Collection<Route> routes;
@@ -286,4 +289,13 @@
public void setRoutePolicy(RoutePolicy routePolicy) {
this.routePolicy = routePolicy;
}
+
+ public int getAndIncrement(ProcessorDefinition<?> node) {
+ AtomicInteger count = nodeIndex.get(node);
+ if (count == null) {
+ count = new AtomicInteger();
+ nodeIndex.put(node, count);
+ }
+ return count.getAndIncrement();
+ }
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteNode.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteNode.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteNode.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteNode.java Sat Nov 14 17:47:18 2009
@@ -20,6 +20,7 @@
import org.apache.camel.Expression;
import org.apache.camel.Processor;
import org.apache.camel.RouteNode;
+import org.apache.camel.management.InstrumentationProcessor;
import org.apache.camel.model.ProcessorDefinition;
import org.apache.camel.processor.Traceable;
@@ -57,12 +58,17 @@
return expression.evaluate(exchange, String.class);
}
- if (processor != null) {
- if (processor instanceof Traceable) {
- Traceable trace = (Traceable) processor;
+ Processor target = processor;
+ if (target != null) {
+ // can be wrapped
+ if (target instanceof InstrumentationProcessor) {
+ target = ((InstrumentationProcessor) target).getProcessor();
+ }
+
+ if (target instanceof Traceable) {
+ Traceable trace = (Traceable) target;
return trace.getTraceLabel();
}
- processor.toString();
}
// default then to definition
Added: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultTracedRouteNodes.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultTracedRouteNodes.java?rev=836224&view=auto
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultTracedRouteNodes.java (added)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultTracedRouteNodes.java Sat Nov 14 17:47:18 2009
@@ -0,0 +1,102 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.impl;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Stack;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.RouteNode;
+import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.spi.TracedRouteNodes;
+
+/**
+ * @version $Revision$
+ */
+public class DefaultTracedRouteNodes implements TracedRouteNodes {
+
+ private Stack<List<RouteNode>> routeNodes = new Stack<List<RouteNode>>();
+ private Map<ProcessorDefinition<?>, AtomicInteger> nodeCounter = new HashMap<ProcessorDefinition<?>, AtomicInteger>();
+
+ public DefaultTracedRouteNodes() {
+ // create an empty list to start with
+ routeNodes.push(new ArrayList<RouteNode>());
+ }
+
+ public void addTraced(RouteNode entry) {
+ List<RouteNode> list = routeNodes.peek();
+ if (list == null) {
+ list = new ArrayList<RouteNode>();
+ routeNodes.push(list);
+ }
+ list.add(entry);
+ }
+
+ public RouteNode getLastNode() {
+ List<RouteNode> list = routeNodes.peek();
+ if (list == null || list.isEmpty()) {
+ return null;
+ }
+ return list.get(list.size() - 1);
+ }
+
+ public RouteNode getSecondLastNode() {
+ List<RouteNode> list = routeNodes.peek();
+ if (list == null || list.isEmpty() || list.size() == 1) {
+ return null;
+ }
+ return list.get(list.size() - 2);
+ }
+
+ public List<RouteNode> getNodes() {
+ List<RouteNode> answer = new ArrayList<RouteNode>();
+ for (List<RouteNode> list : routeNodes) {
+ answer.addAll(list);
+ }
+ return Collections.unmodifiableList(answer);
+ }
+
+ public void popBlock() {
+ routeNodes.pop();
+ }
+
+ public void pushBlock() {
+ // push a new block and add the last node as starting point
+ RouteNode last = getLastNode();
+ routeNodes.push(new ArrayList<RouteNode>());
+ if (last != null) {
+ addTraced(last);
+ }
+ }
+
+ public void clear() {
+ routeNodes.clear();
+ }
+
+ public int getAndIncrementCounter(ProcessorDefinition<?> node) {
+ AtomicInteger count = nodeCounter.get(node);
+ if (count == null) {
+ count = new AtomicInteger();
+ nodeCounter.put(node, count);
+ }
+ return count.getAndIncrement();
+ }
+}
Propchange: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultTracedRouteNodes.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultTracedRouteNodes.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultUnitOfWork.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultUnitOfWork.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultUnitOfWork.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultUnitOfWork.java Sat Nov 14 17:47:18 2009
@@ -17,19 +17,14 @@
package org.apache.camel.impl;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
-import org.apache.camel.RouteNode;
import org.apache.camel.Service;
-import org.apache.camel.model.ProcessorDefinition;
import org.apache.camel.spi.Synchronization;
-import org.apache.camel.spi.TraceableUnitOfWork;
+import org.apache.camel.spi.TracedRouteNodes;
+import org.apache.camel.spi.UnitOfWork;
import org.apache.camel.util.EventHelper;
import org.apache.camel.util.UuidGenerator;
import org.apache.commons.logging.Log;
@@ -40,18 +35,18 @@
*
* @version $Revision$
*/
-public class DefaultUnitOfWork implements TraceableUnitOfWork, Service {
+public class DefaultUnitOfWork implements UnitOfWork, Service {
private static final transient Log LOG = LogFactory.getLog(DefaultUnitOfWork.class);
private String id;
private List<Synchronization> synchronizations;
- private List<RouteNode> routeNodes;
- private Map<ProcessorDefinition<?>, AtomicInteger> routeIndex = new HashMap<ProcessorDefinition<?>, AtomicInteger>();
private Message originalInMessage;
+ private final TracedRouteNodes tracedRouteNodes;
public DefaultUnitOfWork(Exchange exchange) {
- // TODO: optimize to only copy original message if enabled to do so in the route
+ tracedRouteNodes = new DefaultTracedRouteNodes();
+ // TODO: optimize to only copy original message if enabled to do so in the route
// special for JmsMessage as it can cause it to loose headers later.
if (exchange.getIn().getClass().getSimpleName().equals("JmsMessage")) {
this.originalInMessage = new DefaultMessage();
@@ -79,10 +74,9 @@
if (synchronizations != null) {
synchronizations.clear();
}
- if (routeNodes != null) {
- routeNodes.clear();
+ if (tracedRouteNodes != null) {
+ tracedRouteNodes.clear();
}
- routeIndex.clear();
originalInMessage = null;
}
@@ -152,41 +146,12 @@
return id;
}
- public void addTraced(RouteNode entry) {
- if (routeNodes == null) {
- routeNodes = new ArrayList<RouteNode>();
- }
- routeNodes.add(entry);
- }
-
- public RouteNode getLastNode() {
- if (routeNodes == null || routeNodes.isEmpty()) {
- return null;
- }
- return routeNodes.get(routeNodes.size() - 1);
- }
-
- public RouteNode getSecondLastNode() {
- if (routeNodes == null || routeNodes.isEmpty() || routeNodes.size() == 1) {
- return null;
- }
- return routeNodes.get(routeNodes.size() - 2);
- }
-
- public List<RouteNode> getNodes() {
- return Collections.unmodifiableList(routeNodes);
- }
-
public Message getOriginalInMessage() {
return originalInMessage;
}
- public int getAndIncrement(ProcessorDefinition<?> node) {
- AtomicInteger count = routeIndex.get(node);
- if (count == null) {
- count = new AtomicInteger();
- routeIndex.put(node, count);
- }
- return count.getAndIncrement();
+ public TracedRouteNodes getTracedRouteNodes() {
+ return tracedRouteNodes;
}
+
}
Copied: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnCompletionRouteNode.java (from r836130, camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteNode.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnCompletionRouteNode.java?p2=camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnCompletionRouteNode.java&p1=camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteNode.java&r1=836130&r2=836224&rev=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteNode.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnCompletionRouteNode.java Sat Nov 14 17:47:18 2009
@@ -17,64 +17,41 @@
package org.apache.camel.impl;
import org.apache.camel.Exchange;
-import org.apache.camel.Expression;
import org.apache.camel.Processor;
import org.apache.camel.RouteNode;
import org.apache.camel.model.ProcessorDefinition;
-import org.apache.camel.processor.Traceable;
/**
* A default implementation of the {@link org.apache.camel.RouteNode}
*
* @version $Revision$
*/
-public class DefaultRouteNode implements RouteNode {
+public class OnCompletionRouteNode implements RouteNode {
- private Expression expression;
- private Processor processor;
- private ProcessorDefinition<?> processorDefinition;
+ private final Exchange exchange;
- public DefaultRouteNode(ProcessorDefinition<?> processorDefinition, Processor processor) {
- this.processor = processor;
- this.processorDefinition = processorDefinition;
- }
-
- public DefaultRouteNode(ProcessorDefinition<?> processorDefinition, Expression expression) {
- this.processorDefinition = processorDefinition;
- this.expression = expression;
+ public OnCompletionRouteNode(Exchange exchange) {
+ this.exchange = exchange;
}
public Processor getProcessor() {
- return processor;
+ return null;
}
public ProcessorDefinition<?> getProcessorDefinition() {
- return processorDefinition;
+ return null;
}
public String getLabel(Exchange exchange) {
- if (expression != null) {
- return expression.evaluate(exchange, String.class);
- }
-
- if (processor != null) {
- if (processor instanceof Traceable) {
- Traceable trace = (Traceable) processor;
- return trace.getTraceLabel();
- }
- processor.toString();
- }
-
- // default then to definition
- return processorDefinition.getLabel();
+ return "OnCompletion[" + exchange.getProperty(Exchange.CORRELATION_ID) + "]";
}
public boolean isAbstract() {
- return processor == null;
+ return true;
}
@Override
public String toString() {
- return "RouteNode[" + processorDefinition + "]";
+ return "OnCompletionRouteNode";
}
-}
+}
\ No newline at end of file
Added: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnExceptionRouteNode.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnExceptionRouteNode.java?rev=836224&view=auto
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnExceptionRouteNode.java (added)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnExceptionRouteNode.java Sat Nov 14 17:47:18 2009
@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.impl;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.RouteNode;
+import org.apache.camel.model.ProcessorDefinition;
+
+/**
+ * A default implementation of the {@link org.apache.camel.RouteNode}
+ *
+ * @version $Revision$
+ */
+public class OnExceptionRouteNode implements RouteNode {
+
+ private final Exchange exchange;
+
+ public OnExceptionRouteNode(Exchange exchange) {
+ this.exchange = exchange;
+ }
+
+ public Processor getProcessor() {
+ return null;
+ }
+
+ public ProcessorDefinition<?> getProcessorDefinition() {
+ return null;
+ }
+
+ public String getLabel(Exchange exchange) {
+ return "OnException[" + exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class).getClass().getSimpleName() + "]";
+ }
+
+ public boolean isAbstract() {
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return "OnExceptionRouteNode";
+ }
+}
\ No newline at end of file
Propchange: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnExceptionRouteNode.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/OnExceptionRouteNode.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java Sat Nov 14 17:47:18 2009
@@ -65,6 +65,7 @@
import org.apache.commons.logging.LogFactory;
import static org.apache.camel.builder.Builder.body;
+
/**
* Base class for processor types that most XML types extend.
*
@@ -152,10 +153,10 @@
if (processor instanceof Channel) {
return processor;
}
- return wrapChannel(routeContext, processor);
+ return wrapChannel(routeContext, processor, null);
}
- protected Processor wrapChannel(RouteContext routeContext, Processor processor) throws Exception {
+ protected Processor wrapChannel(RouteContext routeContext, Processor processor, ProcessorDefinition<?> child) throws Exception {
// put a channel in between this and each output to control the route flow logic
Channel channel = createChannel(routeContext);
channel.setNextProcessor(processor);
@@ -167,12 +168,14 @@
channel.addInterceptStrategy(routeContext.getManagedInterceptStrategy());
}
addInterceptStrategies(routeContext, channel, this.getInterceptStrategies());
-
- // init the channel
- channel.initChannel(this, routeContext);
// must do this ugly cast to avoid compiler error on HP-UX
ProcessorDefinition defn = (ProcessorDefinition) this;
+
+ // set the child before init the channel
+ channel.setChildDefinition(child);
+ channel.initChannel(defn, routeContext);
+
// set the error handler, must be done after init as we can set the error handler as first in the chain
if (defn instanceof TryDefinition || defn instanceof CatchDefinition || defn instanceof FinallyDefinition) {
// do not use error handler for try .. catch .. finally blocks as it will handle errors itself
@@ -259,7 +262,7 @@
continue;
}
- Processor channel = wrapChannel(routeContext, processor);
+ Processor channel = wrapChannel(routeContext, processor, output);
list.add(channel);
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DefaultChannel.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DefaultChannel.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DefaultChannel.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DefaultChannel.java Sat Nov 14 17:47:18 2009
@@ -59,6 +59,7 @@
// the real output to invoke that has been wrapped
private Processor output;
private ProcessorDefinition<?> definition;
+ private ProcessorDefinition<?> childDefinition;
private CamelContext camelContext;
public List<Processor> next() {
@@ -77,7 +78,7 @@
public Processor getOutput() {
// the errorHandler is already decorated with interceptors
- // so it cointain the entire chain of processors, so we can safely use it directly as output
+ // so it contain the entire chain of processors, so we can safely use it directly as output
// if no error handler provided we use the output
return errorHandler != null ? errorHandler : output;
}
@@ -123,6 +124,10 @@
return definition;
}
+ public void setChildDefinition(ProcessorDefinition<?> childDefinition) {
+ this.childDefinition = childDefinition;
+ }
+
@Override
protected void doStart() throws Exception {
ServiceHelper.startServices(errorHandler, output);
@@ -148,7 +153,9 @@
}
// then wrap the output with the tracer
- TraceInterceptor trace = (TraceInterceptor) getOrCreateTracer().wrapProcessorInInterceptors(routeContext.getCamelContext(), outputDefinition, target, null);
+ // the tracer should have the fine grained definition so if a child is set then use it, if not then its the original output used
+ ProcessorDefinition traceDef = childDefinition != null ? childDefinition : outputDefinition;
+ TraceInterceptor trace = (TraceInterceptor) getOrCreateTracer().wrapProcessorInInterceptors(routeContext.getCamelContext(), traceDef, target, null);
// trace interceptor need to have a reference to route context so we at runtime can enable/disable tracing on-the-fly
trace.setRouteContext(routeContext);
target = trace;
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java Sat Nov 14 17:47:18 2009
@@ -36,6 +36,7 @@
import org.apache.camel.Producer;
import org.apache.camel.impl.ServiceSupport;
import org.apache.camel.processor.aggregate.AggregationStrategy;
+import org.apache.camel.spi.TracedRouteNodes;
import org.apache.camel.util.ExchangeHelper;
import org.apache.camel.util.ServiceHelper;
import org.apache.camel.util.concurrent.AtomicExchange;
@@ -118,7 +119,7 @@
}
public String getTraceLabel() {
- return "Multicast";
+ return "multicast";
}
public void process(Exchange exchange) throws Exception {
@@ -162,13 +163,7 @@
return subExchange;
}
- try {
- // set property which endpoint we send to
- setToEndpoint(subExchange, producer);
- producer.process(subExchange);
- } catch (Exception e) {
- subExchange.setException(e);
- }
+ doProcess(producer, subExchange);
// should we stop in case of an exception occurred during processing?
if (stopOnException && subExchange.getException() != null) {
@@ -208,14 +203,7 @@
Exchange subExchange = pair.getExchange();
updateNewExchange(subExchange, total, pairs);
- // process it sequentially
- try {
- // set property which endpoint we send to
- setToEndpoint(subExchange, producer);
- producer.process(subExchange);
- } catch (Exception e) {
- subExchange.setException(e);
- }
+ doProcess(producer, subExchange);
// should we stop in case of an exception occurred during processing?
if (stopOnException && subExchange.getException() != null) {
@@ -237,6 +225,30 @@
}
}
+ private void doProcess(Processor producer, Exchange exchange) {
+ TracedRouteNodes traced = exchange.getUnitOfWork() != null ? exchange.getUnitOfWork().getTracedRouteNodes() : null;
+
+ try {
+ // prepare tracing starting from a new block
+ if (traced != null) {
+ traced.pushBlock();
+ }
+
+ // set property which endpoint we send to
+ setToEndpoint(exchange, producer);
+
+ // let the producer process it
+ producer.process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ } finally {
+ // pop the block so by next round we have the same staring point and thus the tracing looks accurate
+ if (traced != null) {
+ traced.popBlock();
+ }
+ }
+ }
+
/**
* Aggregate the {@link Exchange} with the current result
*
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultTraceEventMessage.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultTraceEventMessage.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultTraceEventMessage.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultTraceEventMessage.java Sat Nov 14 17:47:18 2009
@@ -23,7 +23,7 @@
import org.apache.camel.Message;
import org.apache.camel.RouteNode;
import org.apache.camel.model.ProcessorDefinition;
-import org.apache.camel.spi.TraceableUnitOfWork;
+import org.apache.camel.spi.TracedRouteNodes;
import org.apache.camel.util.MessageHelper;
/**
@@ -87,18 +87,18 @@
}
private static String extractFromNode(Exchange exchange) {
- if (exchange.getUnitOfWork() instanceof TraceableUnitOfWork) {
- TraceableUnitOfWork tuow = (TraceableUnitOfWork) exchange.getUnitOfWork();
- RouteNode last = tuow.getSecondLastNode();
+ if (exchange.getUnitOfWork() != null) {
+ TracedRouteNodes traced = exchange.getUnitOfWork().getTracedRouteNodes();
+ RouteNode last = traced.getSecondLastNode();
return last != null ? last.getLabel(exchange) : null;
}
return null;
}
private static String extractToNode(Exchange exchange) {
- if (exchange.getUnitOfWork() instanceof TraceableUnitOfWork) {
- TraceableUnitOfWork tuow = (TraceableUnitOfWork) exchange.getUnitOfWork();
- RouteNode last = tuow.getLastNode();
+ if (exchange.getUnitOfWork() != null) {
+ TracedRouteNodes traced = exchange.getUnitOfWork().getTracedRouteNodes();
+ RouteNode last = traced.getLastNode();
return last != null ? last.getLabel(exchange) : null;
}
return null;
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultTraceFormatter.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultTraceFormatter.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultTraceFormatter.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultTraceFormatter.java Sat Nov 14 17:47:18 2009
@@ -20,7 +20,7 @@
import org.apache.camel.Message;
import org.apache.camel.RouteNode;
import org.apache.camel.model.ProcessorDefinition;
-import org.apache.camel.spi.TraceableUnitOfWork;
+import org.apache.camel.spi.TracedRouteNodes;
import org.apache.camel.util.MessageHelper;
/**
@@ -267,17 +267,17 @@
// compute from and to
String from = "";
String to = "";
- if (showNode && exchange.getUnitOfWork() instanceof TraceableUnitOfWork) {
- TraceableUnitOfWork tuow = (TraceableUnitOfWork) exchange.getUnitOfWork();
+ if (showNode) {
+ TracedRouteNodes traced = exchange.getUnitOfWork().getTracedRouteNodes();
- RouteNode traceFrom = tuow.getSecondLastNode();
+ RouteNode traceFrom = traced.getSecondLastNode();
if (traceFrom != null) {
from = getNodeMessage(traceFrom, exchange);
} else if (exchange.getFromEndpoint() != null) {
from = "from(" + exchange.getFromEndpoint().getEndpointUri() + ")";
}
- RouteNode traceTo = tuow.getLastNode();
+ RouteNode traceTo = traced.getLastNode();
if (traceTo != null) {
to = getNodeMessage(traceTo, exchange);
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/TraceInterceptor.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/TraceInterceptor.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/TraceInterceptor.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/TraceInterceptor.java Sat Nov 14 17:47:18 2009
@@ -22,21 +22,22 @@
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
-import org.apache.camel.Expression;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.impl.DefaultExchange;
import org.apache.camel.impl.DefaultRouteNode;
+import org.apache.camel.impl.OnCompletionRouteNode;
import org.apache.camel.model.InterceptDefinition;
import org.apache.camel.model.OnCompletionDefinition;
import org.apache.camel.model.OnExceptionDefinition;
import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.model.TryDefinition;
import org.apache.camel.processor.DelegateProcessor;
import org.apache.camel.processor.Logger;
import org.apache.camel.spi.ExchangeFormatter;
import org.apache.camel.spi.InterceptStrategy;
import org.apache.camel.spi.RouteContext;
-import org.apache.camel.spi.TraceableUnitOfWork;
+import org.apache.camel.spi.TracedRouteNodes;
import org.apache.camel.util.IntrospectionSupport;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.ServiceHelper;
@@ -98,28 +99,29 @@
boolean shouldLog = shouldLogNode(node) && shouldLogExchange(exchange);
// whether we should trace it or not, some nodes should be skipped as they are abstract
- // intermedidate steps for instance related to on completion
+ // intermediate steps for instance related to on completion
boolean trace = true;
// okay this is a regular exchange being routed we might need to log and trace
try {
// before
if (shouldLog) {
-
- // register route path taken if TraceableUnitOfWork unit of work
- if (exchange.getUnitOfWork() instanceof TraceableUnitOfWork) {
- TraceableUnitOfWork tuow = (TraceableUnitOfWork) exchange.getUnitOfWork();
-
- if (node instanceof OnExceptionDefinition) {
- // special for on exception so we can see it in the trace logs
- trace = beforeOnException((OnExceptionDefinition) node, tuow, exchange);
- } else if (node instanceof OnCompletionDefinition) {
- // special for on completion so we can see it in the trace logs
- trace = beforeOnCompletion((OnCompletionDefinition) node, tuow, exchange);
+ // traced holds the information about the current traced route path
+ if (exchange.getUnitOfWork() != null) {
+ TracedRouteNodes traced = exchange.getUnitOfWork().getTracedRouteNodes();
+
+ if (node instanceof OnCompletionDefinition || node instanceof OnExceptionDefinition || node instanceof TryDefinition) {
+ // skip any of these as its just a marker definition
+ trace = false;
+ } else if (exchange.getProperty(Exchange.ON_COMPLETION) != null) {
+ // special for on completion tracing
+ traceOnCompletion(traced, exchange);
} else {
// regular so just add it
- tuow.addTraced(new DefaultRouteNode(node, super.getProcessor()));
+ traced.addTraced(new DefaultRouteNode(node, super.getProcessor()));
}
+ } else {
+ LOG.trace("Cannot trace as this Exchange does not have an UnitOfWork: " + exchange);
}
}
@@ -129,15 +131,10 @@
traceExchange(exchange);
}
- // some nodes need extra work to trace it
- if (exchange.getUnitOfWork() instanceof TraceableUnitOfWork) {
- TraceableUnitOfWork tuow = (TraceableUnitOfWork) exchange.getUnitOfWork();
-
- if (node instanceof InterceptDefinition) {
- // special for intercept() as we would like to trace the processor that was intercepted
- // as well, otherwise we only see the intercepted route, but we need the both to be logged/traced
- afterIntercept((InterceptDefinition) node, tuow, exchange);
- }
+ // special for interceptor where we need to keep booking how far we have routed in the intercepted processors
+ if (node.getParent() instanceof InterceptDefinition && exchange.getUnitOfWork() != null) {
+ TracedRouteNodes traced = exchange.getUnitOfWork().getTracedRouteNodes();
+ traceIntercept((InterceptDefinition) node.getParent(), traced, exchange);
}
// process the exchange
@@ -156,6 +153,35 @@
}
}
+ private void traceOnCompletion(TracedRouteNodes traced, Exchange exchange) {
+ // if ON_COMPLETION is not null then we are actually doing the onCompletion routing
+ // add the onCompletion and then the processor that is invoked next
+ if (!(traced.getSecondLastNode() instanceof OnCompletionRouteNode)) {
+ // only and on completion once when we start the on completion routing
+ traced.addTraced(new OnCompletionRouteNode(exchange));
+ // add the next processor as well
+ traced.addTraced(new DefaultRouteNode(node, super.getProcessor()));
+ } else {
+ // regular during on completion so add it
+ traced.addTraced(new DefaultRouteNode(node, super.getProcessor()));
+ }
+ }
+
+ protected void traceIntercept(InterceptDefinition intercept, TracedRouteNodes traced, Exchange exchange) throws Exception {
+ // use the counter to get the index of the intercepted processor to be traced
+ Processor last = intercept.getInterceptedProcessor(traced.getAndIncrementCounter(intercept));
+ if (last != null) {
+ traced.addTraced(new DefaultRouteNode(node, last));
+
+ boolean shouldLog = shouldLogNode(node) && shouldLogExchange(exchange);
+ if (shouldLog) {
+ // log and trace the processor that was intercepted so we can see it
+ logExchange(exchange);
+ traceExchange(exchange);
+ }
+ }
+ }
+
public Object format(Exchange exchange) {
return formatter.format(this, this.getNode(), exchange);
}
@@ -178,103 +204,6 @@
return tracer;
}
- // Implementation methods
- //-------------------------------------------------------------------------
- protected boolean beforeOnException(OnExceptionDefinition onException, TraceableUnitOfWork tuow, Exchange exchange) throws Exception {
- // lets see if this is the first time for this exception
- int index = tuow.getAndIncrement(node);
- if (index == 0) {
- class OnExceptionExpression implements Expression {
- @SuppressWarnings("unchecked")
- public Object evaluate(Exchange exchange, Class type) {
- String label = "OnException";
- if (exchange.getProperty(Exchange.EXCEPTION_CAUGHT) != null) {
- label += "[" + exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class).getClass().getSimpleName() + "]";
- }
- return exchange.getContext().getTypeConverter().convertTo(type, label);
- }
-
- }
- // yes its first time then do some special to log and trace the
- // start of onException
- Expression exp = new OnExceptionExpression();
-
- // add our pesudo node
- tuow.addTraced(new DefaultRouteNode(node, exp));
-
- // log and trace the processor that was onException so we can see immediately
- logExchange(exchange);
- traceExchange(exchange);
- }
-
- // add the processor that is invoked for this onException
- tuow.addTraced(new DefaultRouteNode(node, super.getProcessor()));
- return true;
- }
-
-
- protected boolean beforeOnCompletion(OnCompletionDefinition onCompletion, TraceableUnitOfWork tuow, Exchange exchange) throws Exception {
- // we should only trace when we do the actual onCompletion
- // the problem is that onCompletion is added at the very beginning of a route to be able to
- // add synchronization hoos on unit of work so it knows to invoke the onCompletion when the
- // exchange is done. But in the trace log we want to defer the onCompletion being logged
- // unitl the exchange is actually completed and is doing the onCompletion routing
- // so if the last node is null then we have just started and thus should not trace this node
- boolean answer = tuow.getLastNode() != null;
-
- if (exchange.getProperty(Exchange.ON_COMPLETION) != null) {
- // if ON_COMPLETION is not null then we are actually doing the onCompletion routing
-
- // we should trace the onCompletion route and we want a start log of the onCompletion
- // step so get the index and see if its 0 then we can add our speical log
- int index = tuow.getAndIncrement(node);
- if (index == 0) {
- class OnCompletionExpression implements Expression {
- @SuppressWarnings("unchecked")
- public Object evaluate(Exchange exchange, Class type) {
- String label = "OnCompletion[" + exchange.getProperty(Exchange.CORRELATION_ID) + "]";
- return exchange.getContext().getTypeConverter().convertTo(type, label);
- }
- }
- // yes its first time then do some special to log and trace the start of onCompletion
- Expression exp = new OnCompletionExpression();
- // add the onCompletion and then the processor that is invoked nest
- tuow.addTraced(new DefaultRouteNode(node, exp));
- tuow.addTraced(new DefaultRouteNode(node, super.getProcessor()));
-
- // log and trace so we get the onCompletion -> processor in the log
- logExchange(exchange);
- traceExchange(exchange);
- } else {
- // we are doing the onCompletion but this is after the start so just
- // add the processor and do no special start message
- tuow.addTraced(new DefaultRouteNode(node, super.getProcessor()));
- }
-
- }
-
- return answer;
- }
-
- protected boolean afterIntercept(InterceptDefinition interceptr, TraceableUnitOfWork tuow, Exchange exchange) throws Exception {
- // get the intercepted processor from the definition
- // we need to use the UoW to have its own index of how far we got into the list
- // of intercepted processors the intercept definition holds as the intercept
- // definition is a single object that is shared by concurrent thread being routed
- // so each exchange has its own private counter
- InterceptDefinition intercept = (InterceptDefinition) node;
- Processor last = intercept.getInterceptedProcessor(tuow.getAndIncrement(intercept));
- if (last != null) {
- tuow.addTraced(new DefaultRouteNode(node, last));
-
- // log and trace the processor that was intercepted so we can see it
- logExchange(exchange);
- traceExchange(exchange);
- }
-
- return true;
- }
-
protected void logExchange(Exchange exchange) {
// process the exchange that formats and logs it
logger.process(exchange);
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/spi/RouteContext.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/spi/RouteContext.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/spi/RouteContext.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/spi/RouteContext.java Sat Nov 14 17:47:18 2009
@@ -185,4 +185,17 @@
* @param routePolicy the custom route policy
*/
void setRoutePolicy(RoutePolicy routePolicy);
+
+ /**
+ * A private counter that increments, is used to as book keeping
+ * when building a route based on the model
+ * <p/>
+ * We need this special book keeping be able to assign the correct
+ * {@link org.apache.camel.model.ProcessorDefinition} to the {@link org.apache.camel.Channel}
+ *
+ * @param node the current node
+ * @return the current count
+ */
+ int getAndIncrement(ProcessorDefinition<?> node);
+
}
Copied: camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TracedRouteNodes.java (from r836130, camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TraceableUnitOfWork.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TracedRouteNodes.java?p2=camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TracedRouteNodes.java&p1=camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TraceableUnitOfWork.java&r1=836130&r2=836224&rev=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TraceableUnitOfWork.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TracedRouteNodes.java Sat Nov 14 17:47:18 2009
@@ -22,13 +22,12 @@
import org.apache.camel.model.ProcessorDefinition;
/**
- * A Unit of work that is also traceable with the
- * {@link org.apache.camel.processor.interceptor.TraceInterceptor} so we can trace the excact
- * route path a given {@link org.apache.camel.Exchange} has been processed.
+ * Tracing information used by {@link org.apache.camel.processor.interceptor.TraceInterceptor}
+ * so we can trace the exact route path a given {@link org.apache.camel.Exchange} has been processed.
*
* @version $Revision$
*/
-public interface TraceableUnitOfWork extends UnitOfWork {
+public interface TracedRouteNodes {
/**
* Adds the entry that was intercepted
@@ -39,21 +38,44 @@
/**
* Gets the last node, is <tt>null</tt> if no last exists.
+ *
+ * @return the last node
*/
RouteNode getLastNode();
/**
* Gets the 2nd last node, is <tt>null</tt> if no last exists.
+ *
+ * @return the 2nd last
*/
RouteNode getSecondLastNode();
/**
* Gets the current list of nodes, representing the route path the
* current {@link org.apache.camel.Exchange} has currently taken.
+ *
+ * @return the node path
*/
List<RouteNode> getNodes();
/**
+ * Prepares a new block for tracing.
+ * <p/>
+ * This is needed when you have child block such as a multicast or aggregator
+ */
+ void pushBlock();
+
+ /**
+ * Pops the last block from tracing.
+ */
+ void popBlock();
+
+ /**
+ * Clears all traced information
+ */
+ void clear();
+
+ /**
* A private counter that increments, is used to as book keeping how far this
* exchange have been intercepted by the general intercept().
* <p/>
@@ -63,5 +85,6 @@
* @param node the intercept node
* @return the current count
*/
- int getAndIncrement(ProcessorDefinition<?> node);
+ int getAndIncrementCounter(ProcessorDefinition<?> node);
+
}
Propchange: camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TracedRouteNodes.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/main/java/org/apache/camel/spi/TracedRouteNodes.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/spi/UnitOfWork.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/spi/UnitOfWork.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/spi/UnitOfWork.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/spi/UnitOfWork.java Sat Nov 14 17:47:18 2009
@@ -73,4 +73,11 @@
* @return the original IN {@link Message}
*/
Message getOriginalInMessage();
+
+ /**
+ * Gets tracing information
+ *
+ * @return trace information
+ */
+ TracedRouteNodes getTracedRouteNodes();
}
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/impl/CustomIdFactoryTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/impl/CustomIdFactoryTest.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/impl/CustomIdFactoryTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/impl/CustomIdFactoryTest.java Sat Nov 14 17:47:18 2009
@@ -88,7 +88,7 @@
assertMockEndpointsSatisfied();
// this should take the when path
- assertEquals("#choice4##when2#", ids);
+ assertEquals("#choice6##when3#", ids);
}
/**
@@ -102,7 +102,7 @@
assertMockEndpointsSatisfied();
// this should take the otherwise path
- assertEquals("#choice4##otherwise3#", ids);
+ assertEquals("#choice6##otherwise5#", ids);
}
private class MyDebuggerCheckingId implements InterceptStrategy {
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DefaultErrorHandlerOnExceptionTraceTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DefaultErrorHandlerOnExceptionTraceTest.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DefaultErrorHandlerOnExceptionTraceTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DefaultErrorHandlerOnExceptionTraceTest.java Sat Nov 14 17:47:18 2009
@@ -61,7 +61,7 @@
public void testWithError() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:boom");
mock.expectedMessageCount(1);
- getMockEndpoint("mock:trace").expectedMessageCount(4);
+ getMockEndpoint("mock:trace").expectedMessageCount(3);
template.sendBody("direct:start", "Kabom");
@@ -70,19 +70,15 @@
TraceEventMessage msg1 = getMockEndpoint("mock:trace").getReceivedExchanges().get(0).getIn().getBody(TraceEventMessage.class);
TraceEventMessage msg2 = getMockEndpoint("mock:trace").getReceivedExchanges().get(1).getIn().getBody(TraceEventMessage.class);
TraceEventMessage msg3 = getMockEndpoint("mock:trace").getReceivedExchanges().get(2).getIn().getBody(TraceEventMessage.class);
- TraceEventMessage msg4 = getMockEndpoint("mock:trace").getReceivedExchanges().get(3).getIn().getBody(TraceEventMessage.class);
assertEquals("direct://start", msg1.getFromEndpointUri());
assertEquals("ref:myProcessor", msg1.getToNode());
assertEquals("ref:myProcessor", msg2.getPreviousNode());
- assertEquals("OnException[IllegalArgumentException]", msg2.getToNode());
+ assertEquals("log://boom", msg2.getToNode());
- assertEquals("OnException[IllegalArgumentException]", msg3.getPreviousNode());
- assertEquals("log://boom", msg3.getToNode());
-
- assertEquals("log://boom", msg4.getPreviousNode());
- assertEquals("mock://boom", msg4.getToNode());
+ assertEquals("log://boom", msg3.getPreviousNode());
+ assertEquals("mock://boom", msg3.getToNode());
}
@Override
Copied: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/TraceDelayerTest.java (from r836130, camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DelayerTest.java)
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/TraceDelayerTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/TraceDelayerTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DelayerTest.java&r1=836130&r2=836224&rev=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/DelayerTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/TraceDelayerTest.java Sat Nov 14 17:47:18 2009
@@ -23,44 +23,24 @@
/**
* @version $Revision$
*/
-public class DelayerTest extends ContextTestSupport {
+public class TraceDelayerTest extends ContextTestSupport {
public void testSendingMessageGetsDelayed() throws Exception {
- MockEndpoint resultEndpoint = resolveMandatoryEndpoint("mock:result", MockEndpoint.class);
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
- // do not wait for the first message
- resultEndpoint.expectedMessageCount(0);
- resultEndpoint.setResultWaitTime(500);
- template.sendBodyAndHeader("seda:a", "<hello>world!</hello>", "MyDelay", 1000);
- // we should not receive it as we wait at most 0.5 sec and it take 1 sec to send
- resultEndpoint.assertIsSatisfied();
+ template.sendBody("direct:start", "Hello World");
- // now if we wait a bit longer we should receive the message!
- resultEndpoint.reset();
- resultEndpoint.expectedMessageCount(1);
- resultEndpoint.assertIsSatisfied();
- }
-
- public void testDelayConstant() throws Exception {
- MockEndpoint resultEndpoint = resolveMandatoryEndpoint("mock:result", MockEndpoint.class);
- resultEndpoint.expectedMessageCount(1);
- // should at least take 1 sec to complete
- resultEndpoint.setMinimumResultWaitTime(900);
- template.sendBody("seda:b", "<hello>world!</hello>");
- resultEndpoint.assertIsSatisfied();
+ assertMockEndpointsSatisfied();
}
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
public void configure() {
- // START SNIPPET: ex
- from("seda:a").delay().header("MyDelay").to("mock:result");
- // END SNIPPET: ex
+ context.setTracing(true);
- // START SNIPPET: ex2
- from("seda:b").delay(1000).to("mock:result");
- // END SNIPPET: ex2
+ from("direct:start").delay(1000).to("mock:foo").to("mock:result");
}
};
}
-}
+}
\ No newline at end of file
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/processor/TraceableUnitOfWorkTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/TraceableUnitOfWorkTest.java?rev=836224&r1=836223&r2=836224&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/processor/TraceableUnitOfWorkTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/processor/TraceableUnitOfWorkTest.java Sat Nov 14 17:47:18 2009
@@ -25,7 +25,7 @@
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.JndiRegistry;
import org.apache.camel.management.InstrumentationProcessor;
-import org.apache.camel.spi.TraceableUnitOfWork;
+import org.apache.camel.spi.TracedRouteNodes;
/**
* @version $Revision$
@@ -69,13 +69,12 @@
// START SNIPPET: e2
private class MyErrorProcessor implements Processor {
public void process(Exchange exchange) throws Exception {
- // cast to TraceableUnitOfWork so we can work on the intercepted node path
- TraceableUnitOfWork tuow = (TraceableUnitOfWork) exchange.getUnitOfWork();
+ TracedRouteNodes traced = exchange.getUnitOfWork().getTracedRouteNodes();
// get the list of intercepted nodes
- List<RouteNode> list = tuow.getNodes();
- // get the 3rd last as its the bean
- Processor last = list.get(list.size() - 3).getProcessor();
+ List<RouteNode> list = traced.getNodes();
+ // get the 2rd last as its the bean
+ Processor last = list.get(list.size() - 2).getProcessor();
// wrapped by JMX
if (last instanceof InstrumentationProcessor) {