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 2013/03/08 15:43:10 UTC
svn commit: r1454409 - in /camel/trunk:
camel-core/src/main/java/org/apache/camel/
camel-core/src/main/java/org/apache/camel/api/management/mbean/
camel-core/src/main/java/org/apache/camel/impl/
camel-core/src/main/java/org/apache/camel/management/ cam...
Author: davsclaus
Date: Fri Mar 8 14:43:10 2013
New Revision: 1454409
URL: http://svn.apache.org/r1454409
Log:
CAMEL-6148: BacklogTracer to do message tracing capturing messages in a backlog, which can be accessed on demand using JMX.
Added:
camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/BacklogTracerEventMessage.java
camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedTracerBacklogMBean.java
camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedBacklogTracer.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/BacklogTracer.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/BacklogTracerInterceptor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultBacklogTracerEventMessage.java
camel/trunk/camel-core/src/test/java/org/apache/camel/management/BacklogTracerIdOnAllNodesTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/management/BacklogTracerTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContext.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java
camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementNamingStrategy.java
camel/trunk/camel-core/src/main/java/org/apache/camel/management/ManagedManagementStrategy.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java
camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierExchangeSentParallelTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedCamelContextTracerTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedTracerOptionsTest.java
camel/trunk/components/camel-core-xml/src/main/java/org/apache/camel/core/xml/AbstractCamelContextFactoryBean.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContext.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContext.java?rev=1454409&r1=1454408&r2=1454409&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContext.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContext.java Fri Mar 8 14:43:10 2013
@@ -960,13 +960,30 @@ public interface CamelContext extends Su
* Sets a custom tracer to be used as the default tracer.
* <p/>
* <b>Note:</b> This must be set before any routes are created,
- * changing the defaultTracer for existing routes is not supported.
+ * changing the default tracer for existing routes is not supported.
*
* @param tracer the custom tracer to use as default tracer
*/
void setDefaultTracer(InterceptStrategy tracer);
/**
+ * Gets the default backlog tracer
+ *
+ * @return the default backlog tracer
+ */
+ InterceptStrategy getDefaultBacklogTracer();
+
+ /**
+ * Sets a custom backlog tracer to be used as the default backlog tracer.
+ * <p/>
+ * <b>Note:</b> This must be set before any routes are created,
+ * changing the default backlog tracer for existing routes is not supported.
+ *
+ * @param backlogTracer the custom tracer to use as default backlog tracer
+ */
+ void setDefaultBacklogTracer(InterceptStrategy backlogTracer);
+
+ /**
* Disables using JMX as {@link org.apache.camel.spi.ManagementStrategy}.
*/
void disableJMX();
Added: camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/BacklogTracerEventMessage.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/BacklogTracerEventMessage.java?rev=1454409&view=auto
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/BacklogTracerEventMessage.java (added)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/BacklogTracerEventMessage.java Fri Mar 8 14:43:10 2013
@@ -0,0 +1,49 @@
+/**
+ * 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.api.management.mbean;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * Represents a traced message by the {@link org.apache.camel.processor.interceptor.BacklogTracer}.
+ */
+public interface BacklogTracerEventMessage extends Serializable {
+
+ String ROOT_TAG = "backlogTracerEventMessage";
+ String TIMESTAMP_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSSZ";
+
+ long getUid();
+
+ Date getTimestamp();
+
+ String getToNode();
+
+ String getExchangeId();
+
+ String getMessageAsXml();
+
+ /**
+ * Dumps the event message as XML using the {@link #ROOT_TAG} as root tag.
+ * <p/>
+ * The <tt>timestamp</tt> tag is formatted in the format defined by {@link #TIMESTAMP_FORMAT}
+ *
+ * @return xml representation of this event
+ */
+ String toXml();
+
+}
Added: camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedTracerBacklogMBean.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedTracerBacklogMBean.java?rev=1454409&view=auto
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedTracerBacklogMBean.java (added)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedTracerBacklogMBean.java Fri Mar 8 14:43:10 2013
@@ -0,0 +1,56 @@
+/**
+ * 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.api.management.mbean;
+
+import java.util.List;
+
+import org.apache.camel.api.management.ManagedAttribute;
+import org.apache.camel.api.management.ManagedOperation;
+
+public interface ManagedTracerBacklogMBean {
+
+ @ManagedAttribute(description = "Is tracing enabled")
+ boolean isEnabled();
+
+ @ManagedAttribute(description = "Is tracing enabled")
+ void setEnabled(boolean enabled);
+
+ @ManagedAttribute(description = "Number of traced messages to keep in the backlog (FIFO queue)")
+ int getBacklogSize();
+
+ @ManagedAttribute(description = "Number of traced messages to keep in the backlog (FIFO queue)")
+ void setBacklogSize(int backlogSize);
+
+ @ManagedAttribute(description = "Number of total traced messages")
+ long getTraceCounter();
+
+ @ManagedOperation(description = "Resets the trace counter")
+ void resetTraceCounter();
+
+ @ManagedOperation(description = "Dumps the traced messages for the given node")
+ List<BacklogTracerEventMessage> dumpTracedMessages(String nodeId);
+
+ @ManagedOperation(description = "Dumps the traced messages for the given node in xml format")
+ String dumpTracedMessagesAsXml(String nodeId);
+
+ @ManagedOperation(description = "Dumps the traced messages for all nodes")
+ List<BacklogTracerEventMessage> dumpAllTracedMessages();
+
+ @ManagedOperation(description = "Dumps the traced messages for all nodes in xml format")
+ String dumpAllTracedMessagesAsXml();
+
+}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java?rev=1454409&r1=1454408&r2=1454409&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java Fri Mar 8 14:43:10 2013
@@ -81,6 +81,7 @@ import org.apache.camel.model.ModelCamel
import org.apache.camel.model.RouteDefinition;
import org.apache.camel.model.RouteDefinitionHelper;
import org.apache.camel.model.RoutesDefinition;
+import org.apache.camel.processor.interceptor.BacklogTracer;
import org.apache.camel.processor.interceptor.Debug;
import org.apache.camel.processor.interceptor.Delayer;
import org.apache.camel.processor.interceptor.HandleFault;
@@ -196,6 +197,7 @@ public class DefaultCamelContext extends
private NodeIdFactory nodeIdFactory = new DefaultNodeIdFactory();
private ProcessorFactory processorFactory;
private InterceptStrategy defaultTracer;
+ private InterceptStrategy defaultBacklogTracer;
private InflightRepository inflightRepository = new DefaultInflightRepository();
private final List<RouteStartupOrder> routeStartupOrder = new ArrayList<RouteStartupOrder>();
// start auto assigning route ids using numbering 1000 and upwards
@@ -2370,8 +2372,19 @@ public class DefaultCamelContext extends
return defaultTracer;
}
- public void setDefaultTracer(InterceptStrategy defaultTracer) {
- this.defaultTracer = defaultTracer;
+ public void setDefaultTracer(InterceptStrategy tracer) {
+ this.defaultTracer = tracer;
+ }
+
+ public InterceptStrategy getDefaultBacklogTracer() {
+ if (defaultBacklogTracer == null) {
+ defaultBacklogTracer = new BacklogTracer(this);
+ }
+ return defaultBacklogTracer;
+ }
+
+ public void setDefaultBacklogTracer(InterceptStrategy backlogTracer) {
+ this.defaultBacklogTracer = backlogTracer;
}
public void disableJMX() {
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java?rev=1454409&r1=1454408&r2=1454409&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java Fri Mar 8 14:43:10 2013
@@ -51,6 +51,7 @@ import org.apache.camel.impl.EndpointReg
import org.apache.camel.impl.EventDrivenConsumerRoute;
import org.apache.camel.impl.ProducerCache;
import org.apache.camel.impl.ThrottlingInflightRoutePolicy;
+import org.apache.camel.management.mbean.ManagedBacklogTracer;
import org.apache.camel.management.mbean.ManagedCamelContext;
import org.apache.camel.management.mbean.ManagedConsumerCache;
import org.apache.camel.management.mbean.ManagedEndpoint;
@@ -69,6 +70,7 @@ import org.apache.camel.model.PolicyDefi
import org.apache.camel.model.ProcessorDefinition;
import org.apache.camel.model.ProcessorDefinitionHelper;
import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.processor.interceptor.BacklogTracer;
import org.apache.camel.processor.interceptor.Tracer;
import org.apache.camel.spi.EventNotifier;
import org.apache.camel.spi.LifecycleStrategy;
@@ -110,6 +112,7 @@ public class DefaultManagementLifecycleS
private volatile boolean initialized;
private final Set<String> knowRouteIds = new HashSet<String>();
private final Map<Tracer, ManagedTracer> managedTracers = new HashMap<Tracer, ManagedTracer>();
+ private final Map<BacklogTracer, ManagedBacklogTracer> managedBacklogTracers = new HashMap<BacklogTracer, ManagedBacklogTracer>();
private final Map<ThreadPoolExecutor, Object> managedThreadPools = new HashMap<ThreadPoolExecutor, Object>();
public DefaultManagementLifecycleStrategy() {
@@ -412,6 +415,16 @@ public class DefaultManagementLifecycleS
managedTracers.put(tracer, mt);
}
return mt;
+ } else if (service instanceof BacklogTracer) {
+ // special for backlog tracer
+ BacklogTracer backlogTracer = (BacklogTracer) service;
+ ManagedBacklogTracer mt = managedBacklogTracers.get(backlogTracer);
+ if (mt == null) {
+ mt = new ManagedBacklogTracer(context, backlogTracer);
+ mt.init(getManagementStrategy());
+ managedBacklogTracers.put(backlogTracer, mt);
+ }
+ return mt;
} else if (service instanceof EventNotifier) {
answer = getManagementObjectStrategy().getManagedObjectForEventNotifier(context, (EventNotifier) service);
} else if (service instanceof Producer) {
@@ -872,6 +885,7 @@ public class DefaultManagementLifecycleS
preServices.clear();
wrappedProcessors.clear();
managedTracers.clear();
+ managedBacklogTracers.clear();
managedThreadPools.clear();
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementNamingStrategy.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementNamingStrategy.java?rev=1454409&r1=1454408&r2=1454409&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementNamingStrategy.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementNamingStrategy.java Fri Mar 8 14:43:10 2013
@@ -198,11 +198,14 @@ public class DefaultManagementNamingStra
}
public ObjectName getObjectNameForTracer(CamelContext context, InterceptStrategy tracer) throws MalformedObjectNameException {
+ // use the simple name of the class as the mbean name (eg Tracer, BacklogTracer)
+ String name = tracer.getClass().getSimpleName();
+
StringBuilder buffer = new StringBuilder();
buffer.append(domainName).append(":");
buffer.append(KEY_CONTEXT + "=").append(getContextId(context)).append(",");
buffer.append(KEY_TYPE + "=" + TYPE_TRACER + ",");
- buffer.append(KEY_NAME + "=").append("Tracer");
+ buffer.append(KEY_NAME + "=").append(name);
return createObjectName(buffer);
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/management/ManagedManagementStrategy.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/management/ManagedManagementStrategy.java?rev=1454409&r1=1454408&r2=1454409&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/management/ManagedManagementStrategy.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/management/ManagedManagementStrategy.java Fri Mar 8 14:43:10 2013
@@ -20,6 +20,7 @@ import javax.management.ObjectName;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
+import org.apache.camel.management.mbean.ManagedBacklogTracer;
import org.apache.camel.management.mbean.ManagedCamelContext;
import org.apache.camel.management.mbean.ManagedComponent;
import org.apache.camel.management.mbean.ManagedConsumer;
@@ -112,6 +113,9 @@ public class ManagedManagementStrategy e
} else if (managedObject instanceof ManagedTracer) {
ManagedTracer mt = (ManagedTracer) managedObject;
objectName = getManagementNamingStrategy().getObjectNameForTracer(mt.getContext(), mt.getTracer());
+ } else if (managedObject instanceof ManagedBacklogTracer) {
+ ManagedBacklogTracer mt = (ManagedBacklogTracer) managedObject;
+ objectName = getManagementNamingStrategy().getObjectNameForTracer(mt.getContext(), mt.getBacklogTracer());
} else if (managedObject instanceof ManagedEventNotifier) {
ManagedEventNotifier men = (ManagedEventNotifier) managedObject;
objectName = getManagementNamingStrategy().getObjectNameForEventNotifier(men.getContext(), men.getEventNotifier());
Added: camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedBacklogTracer.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedBacklogTracer.java?rev=1454409&view=auto
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedBacklogTracer.java (added)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedBacklogTracer.java Fri Mar 8 14:43:10 2013
@@ -0,0 +1,96 @@
+/**
+ * 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.management.mbean;
+
+import java.util.List;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.api.management.ManagedResource;
+import org.apache.camel.api.management.mbean.BacklogTracerEventMessage;
+import org.apache.camel.api.management.mbean.ManagedTracerBacklogMBean;
+import org.apache.camel.processor.interceptor.BacklogTracer;
+import org.apache.camel.spi.ManagementStrategy;
+
+/**
+ * @version
+ */
+@ManagedResource(description = "Managed BacklogTracer")
+public class ManagedBacklogTracer implements ManagedTracerBacklogMBean {
+ private final CamelContext camelContext;
+ private final BacklogTracer backlogTracer;
+
+ public ManagedBacklogTracer(CamelContext camelContext, BacklogTracer backlogTracer) {
+ this.camelContext = camelContext;
+ this.backlogTracer = backlogTracer;
+ }
+
+ public void init(ManagementStrategy strategy) {
+ // do nothing
+ }
+
+ public CamelContext getContext() {
+ return camelContext;
+ }
+
+ public BacklogTracer getBacklogTracer() {
+ return backlogTracer;
+ }
+
+ public boolean getEnabled() {
+ return backlogTracer.isEnabled();
+ }
+
+ public void setEnabled(boolean enabled) {
+ backlogTracer.setEnabled(enabled);
+ }
+
+ public boolean isEnabled() {
+ return backlogTracer.isEnabled();
+ }
+
+ public int getBacklogSize() {
+ return backlogTracer.getBacklogSize();
+ }
+
+ public void setBacklogSize(int backlogSize) {
+ backlogTracer.setBacklogSize(backlogSize);
+ }
+
+ public long getTraceCounter() {
+ return backlogTracer.getTraceCounter();
+ }
+
+ public void resetTraceCounter() {
+ backlogTracer.resetTraceCounter();
+ }
+
+ public List<BacklogTracerEventMessage> dumpTracedMessages(String nodeId) {
+ return backlogTracer.dumpTracedMessages(nodeId);
+ }
+
+ public List<BacklogTracerEventMessage> dumpAllTracedMessages() {
+ return backlogTracer.dumpAllTracedMessages();
+ }
+
+ public String dumpTracedMessagesAsXml(String nodeId) {
+ return backlogTracer.dumpTracedMessagesAsXml(nodeId);
+ }
+
+ public String dumpAllTracedMessagesAsXml() {
+ return backlogTracer.dumpAllTracedMessagesAsXml();
+ }
+}
Added: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/BacklogTracer.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/BacklogTracer.java?rev=1454409&view=auto
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/BacklogTracer.java (added)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/BacklogTracer.java Fri Mar 8 14:43:10 2013
@@ -0,0 +1,216 @@
+/**
+ * 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.interceptor;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Processor;
+import org.apache.camel.api.management.mbean.BacklogTracerEventMessage;
+import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.model.ProcessorDefinitionHelper;
+import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.model.RouteDefinitionHelper;
+import org.apache.camel.spi.InterceptStrategy;
+import org.apache.camel.spi.NodeIdFactory;
+import org.apache.camel.support.ServiceSupport;
+
+/**
+ * A tracer used for message tracing, storing a copy of the message details in a backlog.
+ * <p/>
+ * This tracer allows to store message tracers per node in the Camel routes. The tracers
+ * is stored in a backlog queue (FIFO based) which allows to pull the traced messages on demand.
+ */
+public class BacklogTracer extends ServiceSupport implements InterceptStrategy {
+
+ private final CamelContext camelContext;
+ private boolean enabled;
+ private final AtomicLong traceCounter = new AtomicLong(0);
+ // use a queue with a upper limit to avoid storing too many messages
+ private final Queue<DefaultBacklogTracerEventMessage> queue = new ArrayBlockingQueue<DefaultBacklogTracerEventMessage>(1000);
+ // how many of the last messages per node to keep in the backlog
+ private int backlogSize = 10;
+ // remember the processors we are tracing, which we need later
+ private final Set<ProcessorDefinition<?>> processors = new HashSet<ProcessorDefinition<?>>();
+
+ public BacklogTracer(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ @Override
+ public Processor wrapProcessorInInterceptors(CamelContext context, ProcessorDefinition<?> definition, Processor target, Processor nextTarget) throws Exception {
+ // is this the first output from a route, as we want to know this so we can do special logic in first
+ boolean first = false;
+ RouteDefinition route = ProcessorDefinitionHelper.getRoute(definition);
+ if (route != null && !route.getOutputs().isEmpty()) {
+ first = route.getOutputs().get(0) == definition;
+ }
+
+ processors.add(definition);
+ return new BacklogTracerInterceptor(queue, target, definition, route, first, this);
+ }
+
+ /**
+ * Creates a new backlog tracer.
+ *
+ * @param context Camel context
+ * @return a new backlog tracer
+ */
+ public static BacklogTracer createTracer(CamelContext context) {
+ BacklogTracer tracer = new BacklogTracer(context);
+ return tracer;
+ }
+
+ /**
+ * A helper method to return the BacklogTracer instance if one is enabled
+ *
+ * @return the backlog tracer or null if none can be found
+ */
+ public static BacklogTracer getBacklogTracer(CamelContext context) {
+ List<InterceptStrategy> list = context.getInterceptStrategies();
+ for (InterceptStrategy interceptStrategy : list) {
+ if (interceptStrategy instanceof BacklogTracer) {
+ return (BacklogTracer) interceptStrategy;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Whether or not to trace the given processor definition.
+ *
+ * @param definition the processor definition
+ * @return <tt>true</tt> to trace, <tt>false</tt> to skip tracing
+ */
+ public boolean shouldTrace(ProcessorDefinition<?> definition) {
+ return enabled;
+ }
+
+ public boolean isEnabled() {
+ return enabled;
+ }
+
+ public void setEnabled(boolean enabled) {
+ // okay tracer is enabled then force auto assigning ids
+ if (enabled) {
+ forceAutoAssigningIds();
+ }
+ this.enabled = enabled;
+ }
+
+ public int getBacklogSize() {
+ return backlogSize;
+ }
+
+ public void setBacklogSize(int backlogSize) {
+ if (backlogSize <= 0) {
+ throw new IllegalArgumentException("The backlog size must be a positive number, was: " + backlogSize);
+ }
+ this.backlogSize = backlogSize;
+ }
+
+ public long getTraceCounter() {
+ return traceCounter.get();
+ }
+
+ public void resetTraceCounter() {
+ traceCounter.set(0);
+ }
+
+ public List<BacklogTracerEventMessage> dumpTracedMessages(String nodeId) {
+ List<BacklogTracerEventMessage> answer = new ArrayList<BacklogTracerEventMessage>();
+ if (nodeId != null) {
+ for (DefaultBacklogTracerEventMessage message : queue) {
+ if (nodeId.equals(message.getToNode())) {
+ answer.add(message);
+ }
+ }
+ }
+ return answer;
+ }
+
+ public String dumpTracedMessagesAsXml(String nodeId) {
+ List<BacklogTracerEventMessage> events = dumpTracedMessages(nodeId);
+
+ StringBuilder sb = new StringBuilder();
+ sb.append("<").append(DefaultBacklogTracerEventMessage.ROOT_TAG).append("s>");
+ for (BacklogTracerEventMessage event : events) {
+ sb.append("\n").append(event.toXml());
+ }
+ sb.append("\n</").append(DefaultBacklogTracerEventMessage.ROOT_TAG).append("s>");
+ return sb.toString();
+ }
+
+ public List<BacklogTracerEventMessage> dumpAllTracedMessages() {
+ List<BacklogTracerEventMessage> answer = new ArrayList<BacklogTracerEventMessage>();
+ answer.addAll(queue);
+ queue.clear();
+ return answer;
+ }
+
+ public String dumpAllTracedMessagesAsXml() {
+ List<BacklogTracerEventMessage> events = dumpAllTracedMessages();
+
+ StringBuilder sb = new StringBuilder();
+ sb.append("<").append(BacklogTracerEventMessage.ROOT_TAG).append("s>");
+ for (BacklogTracerEventMessage event : events) {
+ sb.append("\n").append(event.toXml());
+ }
+ sb.append("\n</").append(BacklogTracerEventMessage.ROOT_TAG).append("s>");
+ return sb.toString();
+ }
+
+ long incrementTraceCounter() {
+ return traceCounter.incrementAndGet();
+ }
+
+ void stopProcessor(ProcessorDefinition<?> processorDefinition) {
+ this.processors.remove(processorDefinition);
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ queue.clear();
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ queue.clear();
+ processors.clear();
+ }
+
+ private void forceAutoAssigningIds() {
+ NodeIdFactory factory = camelContext.getNodeIdFactory();
+ if (factory != null) {
+ for (ProcessorDefinition<?> child : processors) {
+ // ensure also the children get ids assigned
+ RouteDefinitionHelper.forceAssignIds(camelContext, child);
+ }
+ }
+ }
+
+}
Added: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/BacklogTracerInterceptor.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/BacklogTracerInterceptor.java?rev=1454409&view=auto
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/BacklogTracerInterceptor.java (added)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/BacklogTracerInterceptor.java Fri Mar 8 14:43:10 2013
@@ -0,0 +1,103 @@
+/**
+ * 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.interceptor;
+
+import java.util.Date;
+import java.util.Queue;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.processor.DelegateAsyncProcessor;
+import org.apache.camel.util.MessageHelper;
+
+/**
+ * An interceptor for tracing messages by using the {@link BacklogTracer}.
+ */
+public class BacklogTracerInterceptor extends DelegateAsyncProcessor {
+
+ private final Queue<DefaultBacklogTracerEventMessage> queue;
+ private final BacklogTracer backlogTracer;
+ private final ProcessorDefinition<?> processorDefinition;
+ private final ProcessorDefinition<?> routeDefinition;
+ private final boolean first;
+
+ public BacklogTracerInterceptor(Queue<DefaultBacklogTracerEventMessage> queue, Processor processor,
+ ProcessorDefinition<?> processorDefinition,
+ ProcessorDefinition<?> routeDefinition, boolean first,
+ BacklogTracer tracer) {
+ super(processor);
+ this.queue = queue;
+ this.processorDefinition = processorDefinition;
+ this.routeDefinition = routeDefinition;
+ this.first = first;
+ this.backlogTracer = tracer;
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ try {
+ if (backlogTracer.shouldTrace(processorDefinition)) {
+ // ensure there is space on the queue
+ int drain = queue.size() - backlogTracer.getBacklogSize();
+ if (drain > 0) {
+ for (int i = 0; i < drain; i++) {
+ queue.poll();
+ }
+ }
+
+ Date timestamp = new Date();
+ String toNode = processorDefinition.getId();
+ String exchangeId = exchange.getExchangeId();
+ String messageAsXml = MessageHelper.dumpAsXml(exchange.getIn());
+
+ // if first we should add a pseudo trace message as well, so we have a starting message (eg from the route)
+ if (first) {
+ Date created = exchange.getProperty(Exchange.CREATED_TIMESTAMP, timestamp, Date.class);
+ String routeId = routeDefinition.getId();
+ DefaultBacklogTracerEventMessage pseudo = new DefaultBacklogTracerEventMessage(backlogTracer.incrementTraceCounter(), created, routeId, exchangeId, messageAsXml);
+ queue.add(pseudo);
+ }
+ DefaultBacklogTracerEventMessage event = new DefaultBacklogTracerEventMessage(backlogTracer.incrementTraceCounter(), timestamp, toNode, exchangeId, messageAsXml);
+ queue.add(event);
+ }
+
+ // invoke processor
+ return super.process(exchange, callback);
+
+ } catch (Exception e) {
+ exchange.setException(e);
+ callback.done(true);
+ return true;
+ }
+ }
+
+ public void stop() throws Exception {
+ super.stop();
+ queue.clear();
+ // notify backlogTracer we are stopping to not leak resources
+ backlogTracer.stopProcessor(processorDefinition);
+ }
+
+ @Override
+ public String toString() {
+ return processor.toString();
+ }
+
+}
+
Added: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultBacklogTracerEventMessage.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultBacklogTracerEventMessage.java?rev=1454409&view=auto
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultBacklogTracerEventMessage.java (added)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultBacklogTracerEventMessage.java Fri Mar 8 14:43:10 2013
@@ -0,0 +1,90 @@
+/**
+ * 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.interceptor;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+import org.apache.camel.api.management.mbean.BacklogTracerEventMessage;
+
+/**
+ * An event message holding the traced message by the {@link BacklogTracer}.
+ */
+public final class DefaultBacklogTracerEventMessage implements BacklogTracerEventMessage {
+
+ private static final long serialVersionUID = 1L;
+
+ private final long uid;
+ private final Date timestamp;
+ private final String toNode;
+ private final String exchangeId;
+ private final String messageAsXml;
+
+ public DefaultBacklogTracerEventMessage(long uid, Date timestamp, String toNode, String exchangeId, String messageAsXml) {
+ this.uid = uid;
+ this.timestamp = timestamp;
+ this.toNode = toNode;
+ this.exchangeId = exchangeId;
+ this.messageAsXml = messageAsXml;
+ }
+
+ public long getUid() {
+ return uid;
+ }
+
+ public Date getTimestamp() {
+ return timestamp;
+ }
+
+ public String getToNode() {
+ return toNode;
+ }
+
+ public String getExchangeId() {
+ return exchangeId;
+ }
+
+ public String getMessageAsXml() {
+ return messageAsXml;
+ }
+
+ @Override
+ public String toString() {
+ return "DefaultBacklogTracerEventMessage[" + exchangeId + " at " + toNode + "]";
+ }
+
+ /**
+ * Dumps the event message as XML using the {@link #ROOT_TAG} as root tag.
+ * <p/>
+ * The <tt>timestamp</tt> tag is formatted in the format defined by {@link #TIMESTAMP_FORMAT}
+ *
+ * @return xml representation of this event
+ */
+ public String toXml() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("<").append(ROOT_TAG).append(">\n");
+ sb.append("<uid>").append(uid).append("</uid>\n");
+ String ts = new SimpleDateFormat(TIMESTAMP_FORMAT).format(timestamp);
+ sb.append("<timestamp>").append(ts).append("</timestamp>\n");
+ sb.append("<toNode>").append(toNode).append("</toNode>\n");
+ sb.append("<exchangeId>").append(exchangeId).append("</exchangeId>\n");
+ sb.append(messageAsXml).append("\n");
+ sb.append("</").append(ROOT_TAG).append(">");
+ return sb.toString();
+ }
+}
+
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java?rev=1454409&r1=1454408&r2=1454409&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java Fri Mar 8 14:43:10 2013
@@ -193,8 +193,9 @@ public class DefaultChannel extends Serv
target = managed.wrapProcessorInInterceptors(routeContext.getCamelContext(), targetOutputDef, target, next);
}
- // then wrap the output with the tracer
- TraceInterceptor trace = (TraceInterceptor) getOrCreateTracer().wrapProcessorInInterceptors(routeContext.getCamelContext(), targetOutputDef, target, null);
+ // then wrap the output with the backlog and tracer (backlog first, as we do not want regular tracer to tracer the backlog)
+ BacklogTracerInterceptor backlog = (BacklogTracerInterceptor) getOrCreateBacklogTracer().wrapProcessorInInterceptors(routeContext.getCamelContext(), targetOutputDef, target, null);
+ TraceInterceptor trace = (TraceInterceptor) getOrCreateTracer().wrapProcessorInInterceptors(routeContext.getCamelContext(), targetOutputDef, backlog, 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;
@@ -291,6 +292,32 @@ public class DefaultChannel extends Serv
return tracer;
}
+ private InterceptStrategy getOrCreateBacklogTracer() {
+ InterceptStrategy tracer = BacklogTracer.getBacklogTracer(camelContext);
+ if (tracer == null) {
+ if (camelContext.getRegistry() != null) {
+ // lookup in registry
+ Map<String, BacklogTracer> map = camelContext.getRegistry().findByTypeWithName(BacklogTracer.class);
+ if (map.size() == 1) {
+ tracer = map.values().iterator().next();
+ }
+ }
+ if (tracer == null) {
+ // fallback to use the default tracer
+ tracer = camelContext.getDefaultBacklogTracer();
+ }
+ }
+
+ // which we must manage as well
+ for (LifecycleStrategy strategy : camelContext.getLifecycleStrategies()) {
+ if (tracer instanceof Service) {
+ strategy.onServiceAdd(camelContext, (Service) tracer, null);
+ }
+ }
+
+ return tracer;
+ }
+
public void process(Exchange exchange) throws Exception {
AsyncProcessorHelper.process(this, exchange);
}
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/management/BacklogTracerIdOnAllNodesTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/management/BacklogTracerIdOnAllNodesTest.java?rev=1454409&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/management/BacklogTracerIdOnAllNodesTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/management/BacklogTracerIdOnAllNodesTest.java Fri Mar 8 14:43:10 2013
@@ -0,0 +1,141 @@
+/**
+ * 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.management;
+
+import java.util.List;
+import javax.management.Attribute;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.api.management.mbean.BacklogTracerEventMessage;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.ChoiceDefinition;
+import org.apache.camel.model.LogDefinition;
+import org.apache.camel.model.OtherwiseDefinition;
+import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.model.ToDefinition;
+import org.apache.camel.model.WhenDefinition;
+
+public class BacklogTracerIdOnAllNodesTest extends ManagementTestSupport {
+
+ @SuppressWarnings("unchecked")
+ public void testBacklogTracerEventMessage() throws Exception {
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on = new ObjectName("org.apache.camel:context=localhost/camel-1,type=tracer,name=BacklogTracer");
+ assertNotNull(on);
+ mbeanServer.isRegistered(on);
+
+ Boolean enabled = (Boolean) mbeanServer.getAttribute(on, "Enabled");
+ assertEquals("Should not be enabled", Boolean.FALSE, enabled);
+
+ Integer size = (Integer) mbeanServer.getAttribute(on, "BacklogSize");
+ assertEquals("Should be 10", 10, size.intValue());
+
+ // enable it
+ mbeanServer.setAttribute(on, new Attribute("Enabled", Boolean.TRUE));
+
+ getMockEndpoint("mock:camel").expectedBodiesReceived("Hello Camel");
+ getMockEndpoint("mock:other").expectedBodiesReceived("Hello World");
+ getMockEndpoint("mock:end").expectedMessageCount(2);
+
+ template.sendBody("direct:start", "Hello Camel");
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+
+ List<Exchange> fooExchanges = getMockEndpoint("mock:foo").getReceivedExchanges();
+ List<Exchange> camelExchanges = getMockEndpoint("mock:camel").getReceivedExchanges();
+
+ RouteDefinition route = context.getRouteDefinitions().get(0);
+ assertNotNull(route);
+
+ ChoiceDefinition choice = (ChoiceDefinition) route.getOutputs().get(0);
+ assertEquals("choice1", choice.getId());
+
+ WhenDefinition when = (WhenDefinition) choice.getOutputs().get(0);
+ assertEquals("when1", when.getId());
+
+ LogDefinition log1 = (LogDefinition) when.getOutputs().get(0);
+ assertEquals("log1", log1.getId());
+
+ ToDefinition to1 = (ToDefinition) when.getOutputs().get(1);
+ assertEquals("camel", to1.getId());
+
+ OtherwiseDefinition other = (OtherwiseDefinition) choice.getOutputs().get(1);
+ assertEquals("otherwise1", other.getId());
+
+ LogDefinition log2 = (LogDefinition) other.getOutputs().get(0);
+ assertEquals("log2", log2.getId());
+
+ ToDefinition to2 = (ToDefinition) other.getOutputs().get(1);
+ assertEquals("to1", to2.getId());
+
+ ToDefinition to3 = (ToDefinition) other.getOutputs().get(2);
+ assertEquals("foo", to3.getId());
+
+ ToDefinition to4 = (ToDefinition) route.getOutputs().get(1);
+ assertEquals("end", to4.getId());
+
+ List<BacklogTracerEventMessage> events = (List<BacklogTracerEventMessage>) mbeanServer.invoke(on, "dumpTracedMessages",
+ new Object[]{"to1"}, new String[]{"java.lang.String"});
+
+ assertNotNull(events);
+ assertEquals(1, events.size());
+
+ BacklogTracerEventMessage event1 = events.get(0);
+ assertEquals("to1", event1.getToNode());
+ assertEquals("<message exchangeId=\"" + fooExchanges.get(0).getExchangeId() + "\">\n"
+ + "<body type=\"java.lang.String\">Hello World</body>\n"
+ + "</message>", event1.getMessageAsXml());
+
+ events = (List<BacklogTracerEventMessage>) mbeanServer.invoke(on, "dumpTracedMessages",
+ new Object[]{"camel"}, new String[]{"java.lang.String"});
+
+ assertNotNull(events);
+ assertEquals(1, events.size());
+
+ event1 = events.get(0);
+ assertEquals("camel", event1.getToNode());
+ assertEquals("<message exchangeId=\"" + camelExchanges.get(0).getExchangeId() + "\">\n"
+ + "<body type=\"java.lang.String\">Hello Camel</body>\n"
+ + "</message>", event1.getMessageAsXml());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.setUseBreadcrumb(false);
+
+ from("direct:start")
+ .choice()
+ .when(body().contains("Camel"))
+ .log("A Camel message")
+ .to("mock:camel").id("camel")
+ .otherwise()
+ .log("Some other kind of message")
+ .to("mock:other") // should auto generate id
+ .to("mock:foo").id("foo")
+ .end()
+ .to("mock:end").id("end");
+ }
+ };
+ }
+
+}
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/management/BacklogTracerTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/management/BacklogTracerTest.java?rev=1454409&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/management/BacklogTracerTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/management/BacklogTracerTest.java Fri Mar 8 14:43:10 2013
@@ -0,0 +1,231 @@
+/**
+ * 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.management;
+
+import java.util.List;
+import javax.management.Attribute;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+import org.w3c.dom.Document;
+import org.w3c.dom.NodeList;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.api.management.mbean.BacklogTracerEventMessage;
+import org.apache.camel.builder.RouteBuilder;
+
+public class BacklogTracerTest extends ManagementTestSupport {
+
+ @SuppressWarnings("unchecked")
+ public void testBacklogTracerEventMessage() throws Exception {
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on = new ObjectName("org.apache.camel:context=localhost/camel-1,type=tracer,name=BacklogTracer");
+ assertNotNull(on);
+ mbeanServer.isRegistered(on);
+
+ Boolean enabled = (Boolean) mbeanServer.getAttribute(on, "Enabled");
+ assertEquals("Should not be enabled", Boolean.FALSE, enabled);
+
+ Integer size = (Integer) mbeanServer.getAttribute(on, "BacklogSize");
+ assertEquals("Should be 10", 10, size.intValue());
+
+ // enable it
+ mbeanServer.setAttribute(on, new Attribute("Enabled", Boolean.TRUE));
+
+ getMockEndpoint("mock:foo").expectedMessageCount(2);
+ getMockEndpoint("mock:bar").expectedMessageCount(2);
+
+ template.sendBody("direct:start", "Hello World");
+ template.sendBody("direct:start", "Bye World");
+
+ assertMockEndpointsSatisfied();
+
+ List<Exchange> exchanges = getMockEndpoint("mock:foo").getReceivedExchanges();
+
+ List<BacklogTracerEventMessage> events = (List<BacklogTracerEventMessage>) mbeanServer.invoke(on, "dumpTracedMessages",
+ new Object[]{"foo"}, new String[]{"java.lang.String"});
+
+ assertNotNull(events);
+ assertEquals(2, events.size());
+
+ BacklogTracerEventMessage event1 = events.get(0);
+ assertEquals("foo", event1.getToNode());
+ assertEquals("<message exchangeId=\"" + exchanges.get(0).getExchangeId() + "\">\n"
+ + "<body type=\"java.lang.String\">Hello World</body>\n"
+ + "</message>", event1.getMessageAsXml());
+
+ BacklogTracerEventMessage event2 = events.get(1);
+ assertEquals("foo", event2.getToNode());
+ assertEquals("<message exchangeId=\"" + exchanges.get(1).getExchangeId() + "\">\n"
+ + "<body type=\"java.lang.String\">Bye World</body>\n"
+ + "</message>", event2.getMessageAsXml());
+ }
+
+ @SuppressWarnings("unchecked")
+ public void testBacklogTracerEventMessageAsXml() throws Exception {
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on = new ObjectName("org.apache.camel:context=localhost/camel-1,type=tracer,name=BacklogTracer");
+ assertNotNull(on);
+ mbeanServer.isRegistered(on);
+
+ Boolean enabled = (Boolean) mbeanServer.getAttribute(on, "Enabled");
+ assertEquals("Should not be enabled", Boolean.FALSE, enabled);
+
+ Integer size = (Integer) mbeanServer.getAttribute(on, "BacklogSize");
+ assertEquals("Should be 10", 10, size.intValue());
+
+ // enable it
+ mbeanServer.setAttribute(on, new Attribute("Enabled", Boolean.TRUE));
+
+ getMockEndpoint("mock:foo").expectedMessageCount(2);
+ getMockEndpoint("mock:bar").expectedMessageCount(2);
+
+ template.sendBody("direct:start", "Hello World");
+ template.sendBody("direct:start", "Bye World");
+
+ assertMockEndpointsSatisfied();
+
+ String events = (String) mbeanServer.invoke(on, "dumpTracedMessagesAsXml",
+ new Object[]{"foo"}, new String[]{"java.lang.String"});
+
+ assertNotNull(events);
+ log.info(events);
+
+ // should be valid XML
+ Document dom = context.getTypeConverter().convertTo(Document.class, events);
+ assertNotNull(dom);
+
+ NodeList list = dom.getElementsByTagName("backlogTracerEventMessage");
+ assertEquals(2, list.getLength());
+ }
+
+ @SuppressWarnings("unchecked")
+ public void testBacklogTracerEventMessageDumpAll() throws Exception {
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on = new ObjectName("org.apache.camel:context=localhost/camel-1,type=tracer,name=BacklogTracer");
+ assertNotNull(on);
+ mbeanServer.isRegistered(on);
+
+ Boolean enabled = (Boolean) mbeanServer.getAttribute(on, "Enabled");
+ assertEquals("Should not be enabled", Boolean.FALSE, enabled);
+
+ // enable it
+ mbeanServer.setAttribute(on, new Attribute("Enabled", Boolean.TRUE));
+
+ getMockEndpoint("mock:foo").expectedMessageCount(2);
+ getMockEndpoint("mock:bar").expectedMessageCount(2);
+
+ template.sendBody("direct:start", "Hello World");
+ template.sendBody("direct:start", "Bye World");
+
+ assertMockEndpointsSatisfied();
+
+ List<Exchange> fooExchanges = getMockEndpoint("mock:foo").getReceivedExchanges();
+ List<Exchange> barExchanges = getMockEndpoint("mock:bar").getReceivedExchanges();
+
+ List<BacklogTracerEventMessage> events = (List<BacklogTracerEventMessage>) mbeanServer.invoke(on, "dumpAllTracedMessages", null, null);
+
+ assertNotNull(events);
+ assertEquals(6, events.size());
+
+ BacklogTracerEventMessage event0 = events.get(0);
+ assertEquals("route1", event0.getToNode());
+ assertEquals("<message exchangeId=\"" + fooExchanges.get(0).getExchangeId() + "\">\n"
+ + "<body type=\"java.lang.String\">Hello World</body>\n"
+ + "</message>", event0.getMessageAsXml());
+
+ BacklogTracerEventMessage event1 = events.get(1);
+ assertEquals("foo", event1.getToNode());
+ assertEquals("<message exchangeId=\"" + fooExchanges.get(0).getExchangeId() + "\">\n"
+ + "<body type=\"java.lang.String\">Hello World</body>\n"
+ + "</message>", event1.getMessageAsXml());
+
+ BacklogTracerEventMessage event2 = events.get(2);
+ assertEquals("bar", event2.getToNode());
+ assertEquals("<message exchangeId=\"" + barExchanges.get(0).getExchangeId() + "\">\n"
+ + "<body type=\"java.lang.String\">Hello World</body>\n"
+ + "</message>", event2.getMessageAsXml());
+
+ BacklogTracerEventMessage event3 = events.get(3);
+ assertEquals("route1", event3.getToNode());
+ assertEquals("<message exchangeId=\"" + fooExchanges.get(1).getExchangeId() + "\">\n"
+ + "<body type=\"java.lang.String\">Bye World</body>\n"
+ + "</message>", event3.getMessageAsXml());
+
+ BacklogTracerEventMessage event4 = events.get(4);
+ assertEquals("foo", event4.getToNode());
+ assertEquals("<message exchangeId=\"" + fooExchanges.get(1).getExchangeId() + "\">\n"
+ + "<body type=\"java.lang.String\">Bye World</body>\n"
+ + "</message>", event3.getMessageAsXml());
+
+ BacklogTracerEventMessage event5 = events.get(5);
+ assertEquals("bar", event5.getToNode());
+ assertEquals("<message exchangeId=\"" + barExchanges.get(1).getExchangeId() + "\">\n"
+ + "<body type=\"java.lang.String\">Bye World</body>\n"
+ + "</message>", event4.getMessageAsXml());
+ }
+
+ @SuppressWarnings("unchecked")
+ public void testBacklogTracerEventMessageDumpAllAsXml() throws Exception {
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on = new ObjectName("org.apache.camel:context=localhost/camel-1,type=tracer,name=BacklogTracer");
+ assertNotNull(on);
+ mbeanServer.isRegistered(on);
+
+ Boolean enabled = (Boolean) mbeanServer.getAttribute(on, "Enabled");
+ assertEquals("Should not be enabled", Boolean.FALSE, enabled);
+
+ // enable it
+ mbeanServer.setAttribute(on, new Attribute("Enabled", Boolean.TRUE));
+
+ getMockEndpoint("mock:foo").expectedMessageCount(2);
+ getMockEndpoint("mock:bar").expectedMessageCount(2);
+
+ template.sendBody("direct:start", "Hello World");
+ template.sendBody("direct:start", "Bye World");
+
+ assertMockEndpointsSatisfied();
+
+ String events = (String) mbeanServer.invoke(on, "dumpAllTracedMessagesAsXml", null, null);
+
+ assertNotNull(events);
+ log.info(events);
+
+ // should be valid XML
+ Document dom = context.getTypeConverter().convertTo(Document.class, events);
+ assertNotNull(dom);
+
+ NodeList list = dom.getElementsByTagName("backlogTracerEventMessage");
+ assertEquals(6, list.getLength());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.setUseBreadcrumb(false);
+
+ from("direct:start")
+ .to("mock:foo").id("foo")
+ .to("mock:bar").id("bar");
+
+ }
+ };
+ }
+
+}
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierExchangeSentParallelTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierExchangeSentParallelTest.java?rev=1454409&r1=1454408&r2=1454409&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierExchangeSentParallelTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/management/EventNotifierExchangeSentParallelTest.java Fri Mar 8 14:43:10 2013
@@ -35,6 +35,9 @@ public class EventNotifierExchangeSentPa
assertMockEndpointsSatisfied();
assertTrue(oneExchangeDone.matchesMockWaitTime());
+ // give little delay due parallel stuff
+ Thread.sleep(100);
+
assertEquals(12, events.size());
// we run parallel so just assert we got 6 sending and 6 sent events
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedCamelContextTracerTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedCamelContextTracerTest.java?rev=1454409&r1=1454408&r2=1454409&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedCamelContextTracerTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedCamelContextTracerTest.java Fri Mar 8 14:43:10 2013
@@ -16,7 +16,6 @@
*/
package org.apache.camel.management;
-import java.util.Set;
import javax.management.Attribute;
import javax.management.MBeanServer;
import javax.management.ObjectName;
@@ -35,9 +34,9 @@ public class ManagedCamelContextTracerTe
MBeanServer mbeanServer = getMBeanServer();
ObjectName camel = ObjectName.getInstance("org.apache.camel:context=localhost/camel-1,type=context,name=\"camel-1\"");
- Set<ObjectName> set = mbeanServer.queryNames(new ObjectName("*:type=tracer,*"), null);
- assertEquals(1, set.size());
- ObjectName on = set.iterator().next();
+ ObjectName on = new ObjectName("org.apache.camel:context=localhost/camel-1,type=tracer,name=Tracer");
+ mbeanServer.isRegistered(camel);
+ mbeanServer.isRegistered(on);
// with tracing
MockEndpoint traced = getMockEndpoint("mock:traced");
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedTracerOptionsTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedTracerOptionsTest.java?rev=1454409&r1=1454408&r2=1454409&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedTracerOptionsTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedTracerOptionsTest.java Fri Mar 8 14:43:10 2013
@@ -16,7 +16,6 @@
*/
package org.apache.camel.management;
-import java.util.Set;
import javax.management.Attribute;
import javax.management.MBeanServer;
import javax.management.ObjectName;
@@ -32,9 +31,8 @@ public class ManagedTracerOptionsTest ex
public void testManagedTracerOptions() throws Exception {
MBeanServer mbeanServer = getMBeanServer();
- Set<ObjectName> set = mbeanServer.queryNames(new ObjectName("*:type=tracer,*"), null);
- assertEquals(1, set.size());
- ObjectName on = set.iterator().next();
+ ObjectName on = new ObjectName("org.apache.camel:context=localhost/camel-1,type=tracer,name=Tracer");
+ mbeanServer.isRegistered(on);
mbeanServer.setAttribute(on, new Attribute("Enabled", Boolean.TRUE));
Boolean enabled = (Boolean) mbeanServer.getAttribute(on, "Enabled");
Modified: camel/trunk/components/camel-core-xml/src/main/java/org/apache/camel/core/xml/AbstractCamelContextFactoryBean.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-core-xml/src/main/java/org/apache/camel/core/xml/AbstractCamelContextFactoryBean.java?rev=1454409&r1=1454408&r2=1454409&view=diff
==============================================================================
--- camel/trunk/components/camel-core-xml/src/main/java/org/apache/camel/core/xml/AbstractCamelContextFactoryBean.java (original)
+++ camel/trunk/components/camel-core-xml/src/main/java/org/apache/camel/core/xml/AbstractCamelContextFactoryBean.java Fri Mar 8 14:43:10 2013
@@ -57,6 +57,7 @@ import org.apache.camel.model.RouteDefin
import org.apache.camel.model.ThreadPoolProfileDefinition;
import org.apache.camel.model.config.PropertiesDefinition;
import org.apache.camel.model.dataformat.DataFormatsDefinition;
+import org.apache.camel.processor.interceptor.BacklogTracer;
import org.apache.camel.processor.interceptor.Delayer;
import org.apache.camel.processor.interceptor.HandleFault;
import org.apache.camel.processor.interceptor.TraceFormatter;
@@ -168,6 +169,11 @@ public abstract class AbstractCamelConte
LOG.info("Using custom Tracer: " + tracer);
getContext().addInterceptStrategy(tracer);
}
+ BacklogTracer backlogTracer = getBeanForType(BacklogTracer.class);
+ if (backlogTracer != null) {
+ LOG.info("Using custom BacklogTracer: " + backlogTracer);
+ getContext().addInterceptStrategy(backlogTracer);
+ }
HandleFault handleFault = getBeanForType(HandleFault.class);
if (handleFault != null) {
LOG.info("Using custom HandleFault: " + handleFault);