You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by js...@apache.org on 2007/07/30 17:04:27 UTC
svn commit: r561006 - in /activemq/camel/trunk: camel-core/
camel-core/src/main/java/org/apache/camel/builder/
camel-core/src/main/java/org/apache/camel/impl/
camel-core/src/main/java/org/apache/camel/model/
camel-core/src/main/java/org/apache/camel/pr...
Author: jstrachan
Date: Mon Jul 30 08:04:26 2007
New Revision: 561006
URL: http://svn.apache.org/viewvc?view=rev&rev=561006
Log:
migrated the DSL code to use the JAXB2 POJOs to enhance reuse across the Java DSL and the XML DSL
Added:
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/MulticastType.java (with props)
Modified:
activemq/camel/trunk/camel-core/pom.xml
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/builder/RouteBuilder.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/RouteContext.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ChoiceType.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ExpressionNode.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/FromType.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/OutputType.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/PolicyRef.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorType.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RouteType.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RoutesType.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ThrottlerType.java
activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Pipeline.java
activemq/camel/trunk/camel-core/src/main/resources/org/apache/camel/model/jaxb.index
activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/ErrorHandlerTest.java
activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/InterceptorBuilderTest.java
activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/RouteBuilderTest.java
activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/MulticastTest.java
activemq/camel/trunk/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailMessage.java
activemq/camel/trunk/components/camel-mail/src/test/java/org/apache/camel/component/mail/MultipleDestinationConsumeTest.java
Modified: activemq/camel/trunk/camel-core/pom.xml
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/pom.xml?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/pom.xml (original)
+++ activemq/camel/trunk/camel-core/pom.xml Mon Jul 30 08:04:26 2007
@@ -40,7 +40,6 @@
<dependency>
<groupId>javax.xml</groupId>
<artifactId>jaxb-api</artifactId>
- <optional>true</optional>
</dependency>
<!-- testing -->
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/builder/RouteBuilder.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/builder/RouteBuilder.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/builder/RouteBuilder.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/builder/RouteBuilder.java Mon Jul 30 08:04:26 2007
@@ -20,6 +20,8 @@
import org.apache.camel.Endpoint;
import org.apache.camel.Route;
import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.model.RoutesType;
+import org.apache.camel.model.RouteType;
import org.apache.camel.processor.DelegateProcessor;
import java.util.ArrayList;
@@ -33,10 +35,9 @@
* @version $Revision$
*/
public abstract class RouteBuilder extends BuilderSupport {
- private List<FromBuilder> fromBuilders = new ArrayList<FromBuilder>();
private AtomicBoolean initalized = new AtomicBoolean(false);
+ private RoutesType routeCollection = new RoutesType();
private List<Route> routes = new ArrayList<Route>();
- private List<DelegateProcessor> interceptors = new ArrayList<DelegateProcessor>();
protected RouteBuilder() {
this(null);
@@ -52,16 +53,13 @@
public abstract void configure() throws Exception;
- public FromBuilder from(String uri) {
- Endpoint endpoint = endpoint(uri);
- return from(endpoint);
+ public RouteType from(String uri) {
+ return routeCollection.from(uri);
}
- public FromBuilder from(Endpoint endpoint) {
- FromBuilder answer = new FromBuilder(this, endpoint);
- addFromBuilder(answer);
- return answer;
+ public RouteType from(Endpoint endpoint) {
+ return routeCollection.from(endpoint);
}
/**
@@ -82,12 +80,12 @@
* @return the current builder
*/
public RouteBuilder inheritErrorHandler(boolean value) {
- setInheritErrorHandler(value);
+ routeCollection.setInheritErrorHandlerFlag(value);
return this;
}
public RouteBuilder intercept(DelegateProcessor interceptor) {
- interceptors.add(interceptor);
+ routeCollection.intercept(interceptor);
return this;
}
@@ -110,24 +108,9 @@
return routes;
}
- /**
- * Returns the builders which have been created
- */
- public List<FromBuilder> getFromBuilders() throws Exception {
- checkInitialized();
- return fromBuilders;
- }
-
- public List<DelegateProcessor> getInterceptors() {
- return interceptors;
- }
// Implementation methods
//-----------------------------------------------------------------------
- public void addFromBuilder(FromBuilder answer) {
- fromBuilders.add(answer);
- }
-
protected void checkInitialized() throws Exception {
if (initalized.compareAndSet(false, true)) {
configure();
@@ -136,10 +119,12 @@
}
protected void populateRoutes(List<Route> routes) throws Exception {
- for (FromBuilder builder : fromBuilders) {
- Route route = builder.createRoute();
- routes.add(route);
+ CamelContext camelContext = getContext();
+ if (camelContext == null) {
+ throw new IllegalArgumentException("No CamelContext has been injected!");
}
+ routeCollection.setCamelContext(camelContext);
+ routeCollection.populateRoutes(routes);
}
/**
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/RouteContext.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/RouteContext.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/RouteContext.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/RouteContext.java Mon Jul 30 08:04:26 2007
@@ -24,7 +24,7 @@
import org.apache.camel.model.FromType;
import org.apache.camel.model.ProcessorType;
import org.apache.camel.model.RouteType;
-import org.apache.camel.processor.MulticastProcessor;
+import org.apache.camel.processor.Pipeline;
import java.util.ArrayList;
import java.util.Collection;
@@ -68,33 +68,7 @@
}
public Processor createProcessor(ProcessorType node) throws Exception {
- Processor processor = createProcessor(node.getOutputs());
- return node.wrapProcessor(this, processor);
- }
-
- public Processor createProcessor(Collection<ProcessorType> outputs) throws Exception {
- List<Processor> list = new ArrayList<Processor>();
- for (ProcessorType output : outputs) {
- Processor processor = output.createProcessor(this);
- list.add(processor);
- }
- return createCompositeProcessor(list);
- }
-
- protected Processor createCompositeProcessor(List<Processor> list) {
- if (list.size() == 0) {
- return null;
- }
- Processor processor;
- if (list.size() == 1) {
- processor = list.get(0);
- }
- else {
- //processor = new CompositeProcessor(list);
- // TODO move into the node itself
- processor = new MulticastProcessor(list);
- }
- return processor;
+ return node.createOutputsProcessor(this);
}
public Endpoint resolveEndpoint(String uri) {
@@ -140,7 +114,7 @@
public void commit() {
// now lets turn all of the event driven consumer processors into a single route
if (!eventDrivenProcessors.isEmpty()) {
- Processor processor = createCompositeProcessor(eventDrivenProcessors);
+ Processor processor = Pipeline.newInstance(eventDrivenProcessors);
routes.add(new EventDrivenConsumerRoute(getEndpoint(), processor));
}
}
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ChoiceType.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ChoiceType.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ChoiceType.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ChoiceType.java Mon Jul 30 08:04:26 2007
@@ -17,7 +17,9 @@
*/
package org.apache.camel.model;
+import org.apache.camel.Predicate;
import org.apache.camel.Processor;
+import org.apache.camel.Endpoint;
import org.apache.camel.impl.RouteContext;
import org.apache.camel.processor.ChoiceProcessor;
import org.apache.camel.processor.FilterProcessor;
@@ -30,6 +32,7 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Collection;
/**
* @version $Revision: 1.1 $
@@ -62,6 +65,46 @@
return new ChoiceProcessor(filters, otherwiseProcessor);
}
+ // Fluent API
+ //-------------------------------------------------------------------------
+ public ChoiceType when(Predicate predicate) {
+ getWhenClauses().add(new WhenType(predicate));
+ return this;
+ }
+
+ public OtherwiseType otherwise() {
+ OtherwiseType answer = new OtherwiseType();
+ setOtherwise(answer);
+ return answer;
+ }
+
+ public ChoiceType to(Endpoint endpoint) {
+ super.to(endpoint);
+ return this;
+ }
+
+ public ChoiceType to(Collection<Endpoint> endpoints) {
+ super.to(endpoints);
+ return this;
+ }
+
+ public ChoiceType to(Endpoint... endpoints) {
+ super.to(endpoints);
+ return this;
+ }
+
+ public ChoiceType to(String uri) {
+ super.to(uri);
+ return this;
+ }
+
+ public ChoiceType to(String... uris) {
+ super.to(uris);
+ return this;
+ }
+
+ // Properties
+ //-------------------------------------------------------------------------
public List<WhenType> getWhenClauses() {
return whenClauses;
}
@@ -71,7 +114,16 @@
}
public List<ProcessorType> getOutputs() {
- return Collections.EMPTY_LIST;
+ if (otherwise != null) {
+ return otherwise.getOutputs();
+ }
+ else if (whenClauses.isEmpty()) {
+ return Collections.EMPTY_LIST;
+ }
+ else {
+ WhenType when = whenClauses.get(whenClauses.size() - 1);
+ return when.getOutputs();
+ }
}
public OtherwiseType getOtherwise() {
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ExpressionNode.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ExpressionNode.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ExpressionNode.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ExpressionNode.java Mon Jul 30 08:04:26 2007
@@ -87,4 +87,12 @@
Processor childProcessor = routeContext.createProcessor(this);
return new FilterProcessor(getExpression().createPredicate(routeContext), childProcessor);
}
+
+
+ @Override
+ protected void configureChild(ProcessorType output) {
+ if (isInheritErrorHandler()) {
+ output.setErrorHandlerBuilder(getErrorHandlerBuilder());
+ }
+ }
}
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/FromType.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/FromType.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/FromType.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/FromType.java Mon Jul 30 08:04:26 2007
@@ -47,6 +47,10 @@
setUri(uri);
}
+ public FromType(Endpoint endpoint) {
+ this.endpoint = endpoint;
+ }
+
@Override
public String toString() {
return "From[" + description(getUri(), getRef(), getEndpoint()) + "]";
Added: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/MulticastType.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/MulticastType.java?view=auto&rev=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/MulticastType.java (added)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/MulticastType.java Mon Jul 30 08:04:26 2007
@@ -0,0 +1,50 @@
+/**
+ *
+ * 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.model;
+
+import org.apache.camel.Processor;
+import org.apache.camel.processor.MulticastProcessor;
+import org.apache.camel.impl.RouteContext;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlAttribute;
+import javax.xml.bind.annotation.XmlRootElement;
+import javax.xml.bind.annotation.XmlTransient;
+import java.util.List;
+
+/**
+ * @version $Revision: 1.1 $
+ */
+@XmlRootElement(name = "multicast")
+@XmlAccessorType(XmlAccessType.FIELD)
+public class MulticastType extends OutputType {
+ @Override
+ public String toString() {
+ return "Multicast[" + getOutputs() + "]";
+ }
+
+ @Override
+ public Processor createProcessor(RouteContext routeContext) throws Exception {
+ return createOutputsProcessor(routeContext);
+ }
+
+ protected Processor createCompositeProcessor(List<Processor> list) {
+ return new MulticastProcessor(list);
+ }
+}
\ No newline at end of file
Propchange: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/MulticastType.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/OutputType.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/OutputType.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/OutputType.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/OutputType.java Mon Jul 30 08:04:26 2007
@@ -17,6 +17,9 @@
*/
package org.apache.camel.model;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlElementRef;
import javax.xml.bind.annotation.XmlType;
@@ -30,6 +33,8 @@
*/
@XmlType(name = "outputType")
public abstract class OutputType extends ProcessorType {
+ private static final transient Log log = LogFactory.getLog(OutputType.class);
+
protected List<ProcessorType> outputs = new ArrayList<ProcessorType>();
private List<InterceptorRef> interceptors = new ArrayList<InterceptorRef>();
@@ -56,9 +61,20 @@
this.interceptors = interceptors;
}
+ @Override
protected void configureChild(ProcessorType output) {
- if (output.isInheritErrorHandler()) {
+ if (isInheritErrorHandler()) {
output.setErrorHandlerBuilder(getErrorHandlerBuilder());
}
+ // don't inherit interceptors by default
+/*
+ List<InterceptorRef> list = output.getInterceptors();
+ if (list == null) {
+ log.warn("No interceptor collection: " + output);
+ }
+ else {
+ list.addAll(getInterceptors());
+ }
+*/
}
}
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/PolicyRef.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/PolicyRef.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/PolicyRef.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/PolicyRef.java Mon Jul 30 08:04:26 2007
@@ -60,7 +60,7 @@
@Override
public Processor createProcessor(RouteContext routeContext) throws Exception {
- Processor childProcessor = super.createProcessor(routeContext);
+ Processor childProcessor = createOutputsProcessor(routeContext);
Policy policy = resolvePolicy(routeContext);
if (policy == null) {
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorType.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorType.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorType.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorType.java Mon Jul 30 08:04:26 2007
@@ -17,30 +17,47 @@
*/
package org.apache.camel.model;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.Predicate;
import org.apache.camel.Processor;
import org.apache.camel.Route;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.builder.Builder;
import org.apache.camel.builder.DeadLetterChannelBuilder;
import org.apache.camel.builder.ErrorHandlerBuilder;
import org.apache.camel.builder.FromBuilder;
import org.apache.camel.builder.IdempotentConsumerBuilder;
import org.apache.camel.builder.NoErrorHandlerBuilder;
+import org.apache.camel.builder.ProcessorBuilder;
import org.apache.camel.converter.ObjectConverter;
-import org.apache.camel.impl.EventDrivenConsumerRoute;
import org.apache.camel.impl.RouteContext;
import org.apache.camel.model.language.ExpressionType;
import org.apache.camel.model.language.LanguageExpression;
import org.apache.camel.processor.DelegateProcessor;
+import org.apache.camel.processor.MulticastProcessor;
+import org.apache.camel.processor.Pipeline;
+import org.apache.camel.processor.RecipientList;
+import org.apache.camel.processor.aggregate.AggregationStrategy;
+import org.apache.camel.processor.idempotent.IdempotentConsumer;
+import org.apache.camel.processor.idempotent.MessageIdRepository;
+import org.apache.camel.spi.Policy;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import javax.xml.bind.annotation.XmlAttribute;
import javax.xml.bind.annotation.XmlTransient;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
/**
* @version $Revision: 1.1 $
*/
public abstract class ProcessorType {
+ public static final String DEFAULT_TRACE_CATEGORY = "org.apache.camel.TRACE";
private ErrorHandlerBuilder errorHandlerBuilder;
private Boolean inheritErrorHandlerFlag = Boolean.TRUE; // TODO not sure how else to use an optional attribute in JAXB2
@@ -48,6 +65,317 @@
public abstract List<InterceptorRef> getInterceptors();
+ public Processor createProcessor(RouteContext routeContext) throws Exception {
+ throw new UnsupportedOperationException("Not implemented yet for class: " + getClass().getName());
+ }
+
+ public Processor createOutputsProcessor(RouteContext routeContext) throws Exception {
+ Collection<ProcessorType> outputs = getOutputs();
+ List<Processor> list = new ArrayList<Processor>();
+ for (ProcessorType output : outputs) {
+ Processor processor = output.createProcessor(routeContext);
+ list.add(processor);
+ }
+ Processor processor = null;
+ if (!list.isEmpty()) {
+ if (list.size() == 1) {
+ processor = list.get(0);
+ }
+ else {
+ processor = createCompositeProcessor(list);
+ }
+ }
+ return processor;
+ }
+
+ public void addRoutes(RouteContext routeContext, Collection<Route> routes) throws Exception {
+ Processor processor = makeProcessor(routeContext);
+ routeContext.addEventDrivenProcessor(processor);
+ }
+
+ /**
+ * Wraps the child processor in whatever necessary interceptors and error handlers
+ */
+ public Processor wrapProcessor(RouteContext routeContext, Processor processor) throws Exception {
+ processor = wrapProcessorInInterceptors(routeContext, processor);
+ return wrapInErrorHandler(processor);
+ }
+
+ // Fluent API
+ //-------------------------------------------------------------------------
+
+ /**
+ * Sends the exchange to the given endpoint URI
+ */
+ public ProcessorType to(String uri) {
+ addOutput(new ToType(uri));
+ return this;
+ }
+
+ /**
+ * Sends the exchange to the given endpoint
+ */
+ public ProcessorType to(Endpoint endpoint) {
+ addOutput(new ToType(endpoint));
+ return this;
+ }
+
+ /**
+ * Sends the exchange to a list of endpoints using the {@link MulticastProcessor} pattern
+ */
+ public ProcessorType to(String... uris) {
+ for (String uri : uris) {
+ addOutput(new ToType(uri));
+ }
+ return this;
+ }
+
+ /**
+ * Sends the exchange to a list of endpoints using the {@link MulticastProcessor} pattern
+ */
+ public ProcessorType to(Endpoint... endpoints) {
+ for (Endpoint endpoint : endpoints) {
+ addOutput(new ToType(endpoint));
+ }
+ return this;
+ }
+
+ /**
+ * Sends the exchange to a list of endpoint using the {@link MulticastProcessor} pattern
+ */
+ public ProcessorType to(Collection<Endpoint> endpoints) {
+ for (Endpoint endpoint : endpoints) {
+ addOutput(new ToType(endpoint));
+ }
+ return this;
+ }
+
+ /**
+ * Multicasts messages to all its child outputs; so that each processor and destination gets a copy of the original message
+ * to avoid the processors interfering with each other.
+ */
+ public MulticastType multicast() {
+ MulticastType answer = new MulticastType();
+ addOutput(answer);
+ return answer;
+ }
+
+ /**
+ * Creates a {@link Pipeline} of the list of endpoints so that the message will get processed by each endpoint in turn
+ * and for request/response the output of one endpoint will be the input of the next endpoint
+ */
+ public ProcessorType pipeline(String... uris) {
+ // TODO pipeline v mulicast
+ return to(uris);
+ }
+
+ /**
+ * Creates a {@link Pipeline} of the list of endpoints so that the message will get processed by each endpoint in turn
+ * and for request/response the output of one endpoint will be the input of the next endpoint
+ */
+ public ProcessorType pipeline(Endpoint... endpoints) {
+ // TODO pipeline v mulicast
+ return to(endpoints);
+ }
+
+ /**
+ * Creates a {@link Pipeline} of the list of endpoints so that the message will get processed by each endpoint in turn
+ * and for request/response the output of one endpoint will be the input of the next endpoint
+ */
+ public ProcessorType pipeline(Collection<Endpoint> endpoints) {
+ // TODO pipeline v mulicast
+ return to(endpoints);
+ }
+
+ /**
+ * Creates an {@link IdempotentConsumer} to avoid duplicate messages
+ */
+ public IdempotentConsumerType idempotentConsumer(Expression messageIdExpression, MessageIdRepository messageIdRepository) {
+ IdempotentConsumerType answer = new IdempotentConsumerType(messageIdExpression, messageIdRepository);
+ addOutput(answer);
+ return answer;
+ }
+
+ /**
+ * Creates a predicate which is applied and only if it is true then
+ * the exchange is forwarded to the destination
+ *
+ * @return the builder for a predicate
+ */
+ public FilterType filter(Predicate predicate) {
+ FilterType filter = new FilterType(predicate);
+ addOutput(filter);
+ return filter;
+ }
+
+ /**
+ * Creates a choice of one or more predicates with an otherwise clause
+ *
+ * @return the builder for a choice expression
+ */
+ public ChoiceType choice() {
+ ChoiceType answer = new ChoiceType();
+ addOutput(answer);
+ return answer;
+ }
+
+ /**
+ * Creates a dynamic <a href="http://activemq.apache.org/camel/recipient-list.html">Recipient List</a> pattern.
+ *
+ * @param receipients is the builder of the expression used in the {@link RecipientList} to decide the destinations
+ */
+ public ProcessorType recipientList(Expression receipients) {
+ RecipientListType answer = new RecipientListType(receipients);
+ addOutput(answer);
+ return this;
+ }
+
+ /**
+ * A builder for the <a href="http://activemq.apache.org/camel/splitter.html">Splitter</a> pattern
+ * where an expression is evaluated to iterate through each of the parts of a message and then each part is then send to some endpoint.
+ *
+ * @param receipients the expression on which to split
+ * @return the builder
+ */
+ public SplitterType splitter(Expression receipients) {
+ SplitterType answer = new SplitterType(receipients);
+ addOutput(answer);
+ return answer;
+ }
+
+ /**
+ * A builder for the <a href="http://activemq.apache.org/camel/resequencer.html">Resequencer</a> pattern
+ * where an expression is evaluated to be able to compare the message exchanges to reorder them. e.g. you
+ * may wish to sort by some header
+ *
+ * @param expression the expression on which to compare messages in order
+ * @return the builder
+ */
+ public ResequencerType resequencer(Expression<Exchange> expression) {
+ return resequencer(Collections.<Expression>singletonList(expression));
+ }
+
+ /**
+ * A builder for the <a href="http://activemq.apache.org/camel/resequencer.html">Resequencer</a> pattern
+ * where a list of expressions are evaluated to be able to compare the message exchanges to reorder them. e.g. you
+ * may wish to sort by some headers
+ *
+ * @param expressions the expressions on which to compare messages in order
+ * @return the builder
+ */
+ public ResequencerType resequencer(List<Expression> expressions) {
+ ResequencerType answer = new ResequencerType(expressions);
+ addOutput(answer);
+ return answer;
+ }
+
+ /**
+ * A builder for the <a href="http://activemq.apache.org/camel/resequencer.html">Resequencer</a> pattern
+ * where a list of expressions are evaluated to be able to compare the message exchanges to reorder them. e.g. you
+ * may wish to sort by some headers
+ *
+ * @param expressions the expressions on which to compare messages in order
+ * @return the builder
+ */
+ public ResequencerType resequencer(Expression... expressions) {
+ List<Expression> list = new ArrayList<Expression>();
+ for (Expression expression : expressions) {
+ list.add(expression);
+ }
+ return resequencer(list);
+ }
+
+ /**
+ * A builder for the <a href="http://activemq.apache.org/camel/aggregator.html">Aggregator</a> pattern
+ * where a batch of messages are processed (up to a maximum amount or until some timeout is reached)
+ * and messages for the same correlation key are combined together using some kind of
+ * {@link AggregationStrategy ) (by default the latest message is used) to compress many message exchanges
+ * into a smaller number of exchanges.
+ * <p/>
+ * A good example of this is stock market data; you may be receiving 30,000 messages/second and you may want to
+ * throttle it right down so that multiple messages for the same stock are combined (or just the latest
+ * message is used and older prices are discarded). Another idea is to combine line item messages together
+ * into a single invoice message.
+ *
+ * @param correlationExpression the expression used to calculate the correlation key. For a JMS message this could
+ * be the expression <code>header("JMSDestination")</code> or <code>header("JMSCorrelationID")</code>
+ */
+ public AggregatorType aggregator(Expression correlationExpression) {
+ AggregatorType answer = new AggregatorType(correlationExpression);
+ addOutput(answer);
+ return answer;
+ }
+
+ /**
+ * A builder for the <a href="http://activemq.apache.org/camel/aggregator.html">Aggregator</a> pattern
+ * where a batch of messages are processed (up to a maximum amount or until some timeout is reached)
+ * and messages for the same correlation key are combined together using some kind of
+ * {@link AggregationStrategy ) (by default the latest message is used) to compress many message exchanges
+ * into a smaller number of exchanges.
+ * <p/>
+ * A good example of this is stock market data; you may be receiving 30,000 messages/second and you may want to
+ * throttle it right down so that multiple messages for the same stock are combined (or just the latest
+ * message is used and older prices are discarded). Another idea is to combine line item messages together
+ * into a single invoice message.
+ *
+ * @param correlationExpression the expression used to calculate the correlation key. For a JMS message this could
+ * be the expression <code>header("JMSDestination")</code> or <code>header("JMSCorrelationID")</code>
+ */
+ public AggregatorType aggregator(Expression correlationExpression, AggregationStrategy aggregationStrategy) {
+ AggregatorType answer = new AggregatorType(correlationExpression, aggregationStrategy);
+ addOutput(answer);
+ return answer;
+ }
+
+ /**
+ * A builder for the <a href="http://activemq.apache.org/camel/delayer.html">Delayer</a> pattern
+ * where an expression is used to calculate the time which the message will be dispatched on
+ *
+ * @param processAtExpression an expression to calculate the time at which the messages should be processed
+ * @return the builder
+ */
+ public DelayerType delayer(Expression<Exchange> processAtExpression) {
+ return delayer(processAtExpression, 0L);
+ }
+
+ /**
+ * A builder for the <a href="http://activemq.apache.org/camel/delayer.html">Delayer</a> pattern
+ * where an expression is used to calculate the time which the message will be dispatched on
+ *
+ * @param processAtExpression an expression to calculate the time at which the messages should be processed
+ * @param delay the delay in milliseconds which is added to the processAtExpression to determine the time the
+ * message should be processed
+ * @return the builder
+ */
+ public DelayerType delayer(Expression<Exchange> processAtExpression, long delay) {
+ DelayerType answer = new DelayerType(processAtExpression, delay);
+ addOutput(answer);
+ return answer;
+ }
+
+ /**
+ * A builder for the <a href="http://activemq.apache.org/camel/delayer.html">Delayer</a> pattern
+ * where a fixed amount of milliseconds are used to delay processing of a message exchange
+ *
+ * @param delay the default delay in milliseconds
+ * @return the builder
+ */
+ public DelayerType delayer(long delay) {
+ return delayer(null, delay);
+ }
+
+ /**
+ * A builder for the <a href="http://activemq.apache.org/camel/delayer.html">Delayer</a> pattern
+ * where an expression is used to calculate the time which the message will be dispatched on
+ *
+ * @return the builder
+ */
+ public ThrottlerType throttler(long maximumRequestCount) {
+ ThrottlerType answer = new ThrottlerType(maximumRequestCount);
+ addOutput(answer);
+ return answer;
+ }
+
public ProcessorType interceptor(String ref) {
getInterceptors().add(new InterceptorRef(ref));
return this;
@@ -63,7 +391,7 @@
public FilterType filter(ExpressionType expression) {
FilterType filter = new FilterType();
filter.setExpression(expression);
- getOutputs().add(filter);
+ addOutput(filter);
return filter;
}
@@ -71,28 +399,132 @@
return filter(new LanguageExpression(language, expression));
}
- public ProcessorType to(String uri) {
- ToType to = new ToType();
- to.setUri(uri);
- getOutputs().add(to);
+ /**
+ * Trace logs the exchange before it goes to the next processing step using the {@link #DEFAULT_TRACE_CATEGORY} logging
+ * category.
+ *
+ * @return
+ */
+ public ProcessorType trace() {
+ return trace(DEFAULT_TRACE_CATEGORY);
+ }
+
+ /**
+ * Trace logs the exchange before it goes to the next processing step using the specified logging
+ * category.
+ *
+ * @param category the logging category trace messages will sent to.
+ * @return
+ */
+ public ProcessorType trace(String category) {
+ final Log log = LogFactory.getLog(category);
+ return intercept(new DelegateProcessor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ log.trace(exchange);
+ processNext(exchange);
+ }
+ });
+ }
+
+ public PolicyRef policies() {
+ PolicyRef answer = new PolicyRef();
+ addOutput(answer);
+ return answer;
+ }
+
+ public PolicyRef policy(Policy policy) {
+ PolicyRef answer = new PolicyRef(policy);
+ addOutput(answer);
+ return answer;
+ }
+
+ public ProcessorType intercept(DelegateProcessor interceptor) {
+ getInterceptors().add(new InterceptorRef(interceptor));
return this;
}
- public Processor createProcessor(RouteContext routeContext) throws Exception {
- throw new UnsupportedOperationException("Not implemented yet for class: " + getClass().getName());
+ /**
+ * Installs the given error handler builder
+ *
+ * @param errorHandlerBuilder the error handler to be used by default for all child routes
+ * @return the current builder with the error handler configured
+ */
+ public ProcessorType errorHandler(ErrorHandlerBuilder errorHandlerBuilder) {
+ setErrorHandlerBuilder(errorHandlerBuilder);
+ return this;
}
- public void addRoutes(RouteContext routeContext, Collection<Route> routes) throws Exception {
- Processor processor = makeProcessor(routeContext);
- routeContext.addEventDrivenProcessor(processor);
+ /**
+ * Configures whether or not the error handler is inherited by every processing node (or just the top most one)
+ *
+ * @param condition the falg as to whether error handlers should be inherited or not
+ * @return the current builder
+ */
+ public ProcessorType inheritErrorHandler(boolean condition) {
+ setInheritErrorHandlerFlag(condition);
+ return this;
}
+ // Transformers
+ //-------------------------------------------------------------------------
+
/**
- * Wraps the child processor in whatever necessary interceptors and error handlers
+ * Adds the custom processor to this destination which could be a final destination, or could be a transformation in a pipeline
*/
- public Processor wrapProcessor(RouteContext routeContext, Processor processor) throws Exception {
- processor = wrapProcessorInInterceptors(routeContext, processor);
- return wrapInErrorHandler(processor);
+ public ProcessorType process(Processor processor) {
+ ProcessorRef answer = new ProcessorRef(processor);
+ addOutput(answer);
+ return this;
+ }
+
+ /**
+ * Adds a processor which sets the body on the IN message
+ */
+ public ProcessorType setBody(Expression expression) {
+ return process(ProcessorBuilder.setBody(expression));
+ }
+
+ /**
+ * Adds a processor which sets the body on the OUT message
+ */
+ public ProcessorType setOutBody(Expression expression) {
+ return process(ProcessorBuilder.setOutBody(expression));
+ }
+
+ /**
+ * Adds a processor which sets the header on the IN message
+ */
+ public ProcessorType setHeader(String name, Expression expression) {
+ return process(ProcessorBuilder.setHeader(name, expression));
+ }
+
+ /**
+ * Adds a processor which sets the header on the OUT message
+ */
+ public ProcessorType setOutHeader(String name, Expression expression) {
+ return process(ProcessorBuilder.setOutHeader(name, expression));
+ }
+
+ /**
+ * Adds a processor which sets the exchange property
+ */
+ public ProcessorType setProperty(String name, Expression expression) {
+ return process(ProcessorBuilder.setProperty(name, expression));
+ }
+
+ /**
+ * Converts the IN message body to the specified type
+ */
+ public ProcessorType convertBodyTo(Class type) {
+ return process(ProcessorBuilder.setBody(Builder.body().convertTo(type)));
+ }
+
+ /**
+ * Converts the OUT message body to the specified type
+ */
+ public ProcessorType convertOutBodyTo(Class type) {
+ return process(ProcessorBuilder.setOutBody(Builder.outBody().convertTo(type)));
}
// Properties
@@ -154,7 +586,8 @@
// Interceptors are optional
DelegateProcessor first = null;
DelegateProcessor last = null;
- List<InterceptorRef> interceptors = getInterceptors();
+ List<InterceptorRef> interceptors = new ArrayList<InterceptorRef>(routeContext.getRoute().getInterceptors());
+ interceptors.addAll(getInterceptors());
if (interceptors != null) {
for (InterceptorRef interceptorRef : interceptors) {
DelegateProcessor p = interceptorRef.createInterceptor(routeContext);
@@ -190,5 +623,22 @@
else {
return new NoErrorHandlerBuilder();
}
+ }
+
+ protected void configureChild(ProcessorType output) {
+ }
+
+ protected void addOutput(ProcessorType processorType) {
+ configureChild(processorType);
+ getOutputs().add(processorType);
+ }
+
+ /**
+ * Creates a new instance of some kind of composite processor which defaults
+ * to using a {@link Pipeline} but derived classes could change the behaviour
+ */
+ protected Processor createCompositeProcessor(List<Processor> list) {
+ //return new MulticastProcessor(list);
+ return new Pipeline(list);
}
}
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RouteType.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RouteType.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RouteType.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RouteType.java Mon Jul 30 08:04:26 2007
@@ -19,30 +19,13 @@
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
import org.apache.camel.Endpoint;
-import org.apache.camel.Exchange;
-import org.apache.camel.Expression;
-import org.apache.camel.Predicate;
-import org.apache.camel.Processor;
import org.apache.camel.Route;
-import org.apache.camel.builder.Builder;
-import org.apache.camel.builder.ErrorHandlerBuilder;
-import org.apache.camel.builder.FromBuilder;
-import org.apache.camel.builder.InterceptorBuilder;
-import org.apache.camel.builder.PolicyBuilder;
-import org.apache.camel.builder.ProcessorBuilder;
-import org.apache.camel.builder.ResequencerBuilder;
import org.apache.camel.impl.RouteContext;
-import org.apache.camel.processor.DelegateProcessor;
-import org.apache.camel.processor.MulticastProcessor;
-import org.apache.camel.processor.Pipeline;
-import org.apache.camel.processor.RecipientList;
-import org.apache.camel.processor.aggregate.AggregationStrategy;
-import org.apache.camel.processor.idempotent.IdempotentConsumer;
-import org.apache.camel.processor.idempotent.MessageIdRepository;
-import org.apache.camel.spi.Policy;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlElementRef;
import javax.xml.bind.annotation.XmlRootElement;
@@ -50,7 +33,6 @@
import javax.xml.bind.annotation.XmlType;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.List;
/**
@@ -60,30 +42,49 @@
*/
@XmlRootElement(name = "route")
@XmlType(propOrder = {"interceptors", "inputs", "outputs"})
+@XmlAccessorType(XmlAccessType.FIELD)
public class RouteType extends ProcessorType implements CamelContextAware {
- public static final String DEFAULT_TRACE_CATEGORY = "org.apache.camel.TRACE";
-
private static final transient Log log = LogFactory.getLog(RouteType.class);
- private CamelContext camelContext;
+ @XmlElement(required = false, name = "interceptor")
private List<InterceptorRef> interceptors = new ArrayList<InterceptorRef>();
+ @XmlElementRef
private List<FromType> inputs = new ArrayList<FromType>();
+ @XmlElementRef
private List<ProcessorType> outputs = new ArrayList<ProcessorType>();
+ @XmlTransient
+ private CamelContext camelContext;
+
+ public RouteType() {
+ }
+
+ public RouteType(String uri) {
+ getInputs().add(new FromType(uri));
+ }
+
+ public RouteType(Endpoint endpoint) {
+ getInputs().add(new FromType(endpoint));
+ }
@Override
public String toString() {
return "Route[ " + inputs + " -> " + outputs + "]";
}
+ // TODO should we zap this and replace with next method?
public void addRoutes(CamelContext context) throws Exception {
- setCamelContext(context);
-
Collection<Route> routes = new ArrayList<Route>();
+ addRoutes(context, routes);
+
+ context.addRoutes(routes);
+ }
+
+ public void addRoutes(CamelContext context, Collection<Route> routes) throws Exception {
+ setCamelContext(context);
+
for (FromType fromType : inputs) {
addRoutes(routes, fromType);
}
-
- context.addRoutes(routes);
}
public Endpoint resolveEndpoint(String uri) {
@@ -105,401 +106,9 @@
return this;
}
- /**
- * Sends the exchange to the given endpoint URI
- */
- public RouteType to(String uri) {
- getOutputs().add(new ToType(uri));
- return this;
- }
-
- /**
- * Sends the exchange to the given endpoint
- */
- public RouteType to(Endpoint endpoint) {
- getOutputs().add(new ToType(endpoint));
- return this;
- }
-
- /**
- * Sends the exchange to a list of endpoints using the {@link MulticastProcessor} pattern
- */
- public RouteType to(String... uris) {
- for (String uri : uris) {
- getOutputs().add(new ToType(uri));
- }
- return this;
- }
-
- /**
- * Sends the exchange to a list of endpoints using the {@link MulticastProcessor} pattern
- */
- public RouteType to(Endpoint... endpoints) {
- for (Endpoint endpoint : endpoints) {
- getOutputs().add(new ToType(endpoint));
- }
- return this;
- }
-
- /**
- * Sends the exchange to a list of endpoint using the {@link MulticastProcessor} pattern
- */
- public RouteType to(Collection<Endpoint> endpoints) {
- for (Endpoint endpoint : endpoints) {
- getOutputs().add(new ToType(endpoint));
- }
- return this;
- }
-
- /**
- * Creates a {@link Pipeline} of the list of endpoints so that the message will get processed by each endpoint in turn
- * and for request/response the output of one endpoint will be the input of the next endpoint
- */
- public RouteType pipeline(String... uris) {
- // TODO pipeline v mulicast
- return to(uris);
- }
-
- /**
- * Creates a {@link Pipeline} of the list of endpoints so that the message will get processed by each endpoint in turn
- * and for request/response the output of one endpoint will be the input of the next endpoint
- */
- public RouteType pipeline(Endpoint... endpoints) {
- // TODO pipeline v mulicast
- return to(endpoints);
- }
-
- /**
- * Creates a {@link Pipeline} of the list of endpoints so that the message will get processed by each endpoint in turn
- * and for request/response the output of one endpoint will be the input of the next endpoint
- */
- public RouteType pipeline(Collection<Endpoint> endpoints) {
- // TODO pipeline v mulicast
- return to(endpoints);
- }
-
- /**
- * Creates an {@link IdempotentConsumer} to avoid duplicate messages
- */
- public IdempotentConsumerType idempotentConsumer(Expression messageIdExpression, MessageIdRepository messageIdRepository) {
- IdempotentConsumerType answer = new IdempotentConsumerType(messageIdExpression, messageIdRepository);
- getOutputs().add(answer);
- return answer;
- }
-
- /**
- * Creates a predicate which is applied and only if it is true then
- * the exchange is forwarded to the destination
- *
- * @return the builder for a predicate
- */
- public FilterType filter(Predicate predicate) {
- FilterType filter = new FilterType(predicate);
- getOutputs().add(filter);
- return filter;
- }
-
- /**
- * Creates a choice of one or more predicates with an otherwise clause
- *
- * @return the builder for a choice expression
- */
- public ChoiceType choice() {
- ChoiceType answer = new ChoiceType();
- getOutputs().add(answer);
- return answer;
- }
-
- /**
- * Creates a dynamic <a href="http://activemq.apache.org/camel/recipient-list.html">Recipient List</a> pattern.
- *
- * @param receipients is the builder of the expression used in the {@link RecipientList} to decide the destinations
- */
- public RouteType recipientList(Expression receipients) {
- RecipientListType answer = new RecipientListType(receipients);
- getOutputs().add(answer);
- return this;
- }
-
- /**
- * A builder for the <a href="http://activemq.apache.org/camel/splitter.html">Splitter</a> pattern
- * where an expression is evaluated to iterate through each of the parts of a message and then each part is then send to some endpoint.
- *
- * @param receipients the expression on which to split
- * @return the builder
- */
- public SplitterType splitter(Expression receipients) {
- SplitterType answer = new SplitterType(receipients);
- getOutputs().add(answer);
- return answer;
- }
-
- /**
- * A builder for the <a href="http://activemq.apache.org/camel/resequencer.html">Resequencer</a> pattern
- * where an expression is evaluated to be able to compare the message exchanges to reorder them. e.g. you
- * may wish to sort by some header
- *
- * @param expression the expression on which to compare messages in order
- * @return the builder
- */
- public ResequencerType resequencer(Expression<Exchange> expression) {
- return resequencer(Collections.<Expression>singletonList(expression));
- }
-
- /**
- * A builder for the <a href="http://activemq.apache.org/camel/resequencer.html">Resequencer</a> pattern
- * where a list of expressions are evaluated to be able to compare the message exchanges to reorder them. e.g. you
- * may wish to sort by some headers
- *
- * @param expressions the expressions on which to compare messages in order
- * @return the builder
- */
- public ResequencerType resequencer(List<Expression> expressions) {
- ResequencerType answer = new ResequencerType(expressions);
- getOutputs().add(answer);
- return answer;
- }
-
- /**
- * A builder for the <a href="http://activemq.apache.org/camel/resequencer.html">Resequencer</a> pattern
- * where a list of expressions are evaluated to be able to compare the message exchanges to reorder them. e.g. you
- * may wish to sort by some headers
- *
- * @param expressions the expressions on which to compare messages in order
- * @return the builder
- */
- public ResequencerType resequencer(Expression... expressions) {
- List<Expression> list = new ArrayList<Expression>();
- for (Expression expression : expressions) {
- list.add(expression);
- }
- return resequencer(list);
- }
-
- /**
- * A builder for the <a href="http://activemq.apache.org/camel/aggregator.html">Aggregator</a> pattern
- * where a batch of messages are processed (up to a maximum amount or until some timeout is reached)
- * and messages for the same correlation key are combined together using some kind of
- * {@link AggregationStrategy ) (by default the latest message is used) to compress many message exchanges
- * into a smaller number of exchanges.
- * <p/>
- * A good example of this is stock market data; you may be receiving 30,000 messages/second and you may want to
- * throttle it right down so that multiple messages for the same stock are combined (or just the latest
- * message is used and older prices are discarded). Another idea is to combine line item messages together
- * into a single invoice message.
- *
- * @param correlationExpression the expression used to calculate the correlation key. For a JMS message this could
- * be the expression <code>header("JMSDestination")</code> or <code>header("JMSCorrelationID")</code>
- */
- public AggregatorType aggregator(Expression correlationExpression) {
- AggregatorType answer = new AggregatorType(correlationExpression);
- getOutputs().add(answer);
- return answer;
- }
-
- /**
- * A builder for the <a href="http://activemq.apache.org/camel/aggregator.html">Aggregator</a> pattern
- * where a batch of messages are processed (up to a maximum amount or until some timeout is reached)
- * and messages for the same correlation key are combined together using some kind of
- * {@link AggregationStrategy ) (by default the latest message is used) to compress many message exchanges
- * into a smaller number of exchanges.
- * <p/>
- * A good example of this is stock market data; you may be receiving 30,000 messages/second and you may want to
- * throttle it right down so that multiple messages for the same stock are combined (or just the latest
- * message is used and older prices are discarded). Another idea is to combine line item messages together
- * into a single invoice message.
- *
- * @param correlationExpression the expression used to calculate the correlation key. For a JMS message this could
- * be the expression <code>header("JMSDestination")</code> or <code>header("JMSCorrelationID")</code>
- */
- public AggregatorType aggregator(Expression correlationExpression, AggregationStrategy aggregationStrategy) {
- AggregatorType answer = new AggregatorType(correlationExpression, aggregationStrategy);
- getOutputs().add(answer);
- return answer;
- }
-
- /**
- * A builder for the <a href="http://activemq.apache.org/camel/delayer.html">Delayer</a> pattern
- * where an expression is used to calculate the time which the message will be dispatched on
- *
- * @param processAtExpression an expression to calculate the time at which the messages should be processed
- * @return the builder
- */
- public DelayerType delayer(Expression<Exchange> processAtExpression) {
- return delayer(processAtExpression, 0L);
- }
-
- /**
- * A builder for the <a href="http://activemq.apache.org/camel/delayer.html">Delayer</a> pattern
- * where an expression is used to calculate the time which the message will be dispatched on
- *
- * @param processAtExpression an expression to calculate the time at which the messages should be processed
- * @param delay the delay in milliseconds which is added to the processAtExpression to determine the time the
- * message should be processed
- * @return the builder
- */
- public DelayerType delayer(Expression<Exchange> processAtExpression, long delay) {
- DelayerType answer = new DelayerType(processAtExpression, delay);
- getOutputs().add(answer);
- return answer;
- }
-
- /**
- * A builder for the <a href="http://activemq.apache.org/camel/delayer.html">Delayer</a> pattern
- * where a fixed amount of milliseconds are used to delay processing of a message exchange
- *
- * @param delay the default delay in milliseconds
- * @return the builder
- */
- public DelayerType delayer(long delay) {
- return delayer(null, delay);
- }
-
- /**
- * A builder for the <a href="http://activemq.apache.org/camel/delayer.html">Delayer</a> pattern
- * where an expression is used to calculate the time which the message will be dispatched on
- *
- * @return the builder
- */
- public ThrottlerType throttler(long maximumRequestCount) {
- ThrottlerType answer = new ThrottlerType(maximumRequestCount);
- getOutputs().add(answer);
- return answer;
- }
-
- /**
- * Installs the given error handler builder
- *
- * @param errorHandlerBuilder the error handler to be used by default for all child routes
- * @return the current builder with the error handler configured
- */
- public RouteType errorHandler(ErrorHandlerBuilder errorHandlerBuilder) {
- setErrorHandlerBuilder(errorHandlerBuilder);
- return this;
- }
-
- /**
- * Configures whether or not the error handler is inherited by every processing node (or just the top most one)
- *
- * @param condition the falg as to whether error handlers should be inherited or not
- * @return the current builder
- */
- public RouteType inheritErrorHandler(boolean condition) {
- setInheritErrorHandlerFlag(condition);
- return this;
- }
-
-
- /**
- * Trace logs the exchange before it goes to the next processing step using the {@link #DEFAULT_TRACE_CATEGORY} logging
- * category.
- *
- * @return
- */
- public RouteType trace() {
- return trace(DEFAULT_TRACE_CATEGORY);
- }
-
- /**
- * Trace logs the exchange before it goes to the next processing step using the specified logging
- * category.
- *
- * @param category the logging category trace messages will sent to.
- * @return
- */
- public RouteType trace(String category) {
- final Log log = LogFactory.getLog(category);
- return intercept(new DelegateProcessor() {
- @Override
- public void process(Exchange exchange) throws Exception {
- log.trace(exchange);
- processNext(exchange);
- }
- });
- }
-
- public PolicyRef policies() {
- PolicyRef answer = new PolicyRef();
- getOutputs().add(answer);
- return answer;
- }
-
- public PolicyRef policy(Policy policy) {
- PolicyRef answer = new PolicyRef(policy);
- getOutputs().add(answer);
- return answer;
- }
-
- public RouteType intercept(DelegateProcessor interceptor) {
- getInterceptors().add(new InterceptorRef(interceptor));
- return this;
- }
-
- // Transformers
- //-------------------------------------------------------------------------
-
- /**
- * Adds the custom processor to this destination which could be a final destination, or could be a transformation in a pipeline
- */
- public RouteType process(Processor processor) {
- ProcessorRef answer = new ProcessorRef(processor);
- getOutputs().add(answer);
- return this;
- }
-
- /**
- * Adds a processor which sets the body on the IN message
- */
- public RouteType setBody(Expression expression) {
- return process(ProcessorBuilder.setBody(expression));
- }
-
- /**
- * Adds a processor which sets the body on the OUT message
- */
- public RouteType setOutBody(Expression expression) {
- return process(ProcessorBuilder.setOutBody(expression));
- }
-
- /**
- * Adds a processor which sets the header on the IN message
- */
- public RouteType setHeader(String name, Expression expression) {
- return process(ProcessorBuilder.setHeader(name, expression));
- }
-
- /**
- * Adds a processor which sets the header on the OUT message
- */
- public RouteType setOutHeader(String name, Expression expression) {
- return process(ProcessorBuilder.setOutHeader(name, expression));
- }
-
- /**
- * Adds a processor which sets the exchange property
- */
- public RouteType setProperty(String name, Expression expression) {
- return process(ProcessorBuilder.setProperty(name, expression));
- }
-
- /**
- * Converts the IN message body to the specified type
- */
- public RouteType convertBodyTo(Class type) {
- return process(ProcessorBuilder.setBody(Builder.body().convertTo(type)));
- }
-
- /**
- * Converts the OUT message body to the specified type
- */
- public RouteType convertOutBodyTo(Class type) {
- return process(ProcessorBuilder.setOutBody(Builder.outBody().convertTo(type)));
- }
-
// Properties
//-----------------------------------------------------------------------
- @XmlElement(required = false, name = "interceptor")
public List<InterceptorRef> getInterceptors() {
return interceptors;
}
@@ -508,7 +117,6 @@
this.interceptors = interceptors;
}
- @XmlElementRef
public List<FromType> getInputs() {
return inputs;
}
@@ -517,7 +125,6 @@
this.inputs = inputs;
}
- @XmlElementRef
public List<ProcessorType> getOutputs() {
return outputs;
}
@@ -532,7 +139,6 @@
}
}
- @XmlTransient
public CamelContext getCamelContext() {
return camelContext;
}
@@ -555,7 +161,11 @@
routeContext.commit();
}
+ @Override
protected void configureChild(ProcessorType output) {
+ if (isInheritErrorHandler()) {
+ output.setErrorHandlerBuilder(getErrorHandlerBuilder());
+ }
List<InterceptorRef> list = output.getInterceptors();
if (list == null) {
log.warn("No interceptor collection: " + output);
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RoutesType.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RoutesType.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RoutesType.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RoutesType.java Mon Jul 30 08:04:26 2007
@@ -16,8 +16,17 @@
*/
package org.apache.camel.model;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Route;
+import org.apache.camel.processor.DelegateProcessor;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlAttribute;
import javax.xml.bind.annotation.XmlElementRef;
import javax.xml.bind.annotation.XmlRootElement;
+import javax.xml.bind.annotation.XmlTransient;
import java.util.ArrayList;
import java.util.List;
@@ -27,18 +36,30 @@
* @version $Revision: $
*/
@XmlRootElement(name = "routes")
+@XmlAccessorType(XmlAccessType.FIELD)
public class RoutesType implements RouteContainer {
+ @XmlAttribute
+ private Boolean inheritErrorHandlerFlag = Boolean.TRUE; // TODO not sure how else to use an optional attribute in JAXB2
+ @XmlElementRef
private List<RouteType> routes = new ArrayList<RouteType>();
+ @XmlTransient
+ private List<InterceptorRef> interceptors = new ArrayList<InterceptorRef>();
+ @XmlTransient
+ private CamelContext camelContext;
@Override
public String toString() {
return "Routes: " + routes;
}
+ public void populateRoutes(List<Route> answer) throws Exception {
+ for (RouteType route : routes) {
+ route.addRoutes(camelContext, answer);
+ }
+ }
+
// Properties
//-----------------------------------------------------------------------
-
- @XmlElementRef
public List<RouteType> getRoutes() {
return routes;
}
@@ -47,11 +68,60 @@
this.routes = routes;
}
+ public List<InterceptorRef> getInterceptors() {
+ return interceptors;
+ }
+
+ public void setInterceptors(List<InterceptorRef> interceptors) {
+ this.interceptors = interceptors;
+ }
+
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ public Boolean getInheritErrorHandlerFlag() {
+ return inheritErrorHandlerFlag;
+ }
+
+ public void setInheritErrorHandlerFlag(Boolean inheritErrorHandlerFlag) {
+ this.inheritErrorHandlerFlag = inheritErrorHandlerFlag;
+ }
+
// Fluent API
//-------------------------------------------------------------------------
public RouteType route() {
RouteType route = new RouteType();
+ return route(route);
+ }
+
+ public RouteType from(String uri) {
+ RouteType route = new RouteType(uri);
+ return route(route);
+ }
+
+ public RouteType from(Endpoint endpoint) {
+ RouteType route = new RouteType(endpoint);
+ return route(route);
+ }
+
+ public RouteType route(RouteType route) {
+ // lets configure the route
+ route.setCamelContext(getCamelContext());
+ route.setInheritErrorHandlerFlag(getInheritErrorHandlerFlag());
+ route.getInterceptors().addAll(getInterceptors());
+
getRoutes().add(route);
return route;
}
+
+ public RoutesType intercept(DelegateProcessor interceptor) {
+ getInterceptors().add(new InterceptorRef(interceptor));
+ return this;
+ }
+
}
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ThrottlerType.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ThrottlerType.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ThrottlerType.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ThrottlerType.java Mon Jul 30 08:04:26 2007
@@ -24,9 +24,9 @@
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlAttribute;
+import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlElementRef;
import javax.xml.bind.annotation.XmlRootElement;
-import javax.xml.bind.annotation.XmlElement;
import java.util.ArrayList;
import java.util.List;
@@ -57,6 +57,26 @@
return "Throttler[" + getMaximumRequestsPerPeriod() + " request per " + getTimePeriodMillis() + " millis -> " + getOutputs() + "]";
}
+ @Override
+ public Processor createProcessor(RouteContext routeContext) throws Exception {
+ Processor childProcessor = routeContext.createProcessor(this);
+ return new Throttler(childProcessor, maximumRequestsPerPeriod, timePeriodMillis);
+ }
+
+ // Fluent API
+ //-------------------------------------------------------------------------
+
+ /**
+ * Sets the time period during which the maximum request count is valid for
+ */
+ public ThrottlerType timePeriodMillis(long timePeriodMillis) {
+ this.timePeriodMillis = timePeriodMillis;
+ return this;
+ }
+
+ // Properties
+ //-------------------------------------------------------------------------
+
public Long getMaximumRequestsPerPeriod() {
return maximumRequestsPerPeriod;
}
@@ -87,11 +107,5 @@
public void setOutputs(List<ProcessorType> outputs) {
this.outputs = outputs;
- }
-
- @Override
- public Processor createProcessor(RouteContext routeContext) throws Exception {
- Processor childProcessor = routeContext.createProcessor(this);
- return new Throttler(childProcessor, maximumRequestsPerPeriod, timePeriodMillis);
}
}
Modified: activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Pipeline.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Pipeline.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Pipeline.java (original)
+++ activemq/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Pipeline.java Mon Jul 30 08:04:26 2007
@@ -17,12 +17,11 @@
*/
package org.apache.camel.processor;
-import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
-import org.apache.camel.Producer;
import java.util.Collection;
+import java.util.List;
/**
* Creates a Pipeline pattern where the output of the previous step is sent as input to the next step, reusing the same
@@ -31,6 +30,16 @@
* @version $Revision$
*/
public class Pipeline extends MulticastProcessor implements Processor {
+ public static Processor newInstance(List<Processor> processors) {
+ if (processors.isEmpty()) {
+ return null;
+ }
+ else if (processors.size() == 1) {
+ return processors.get(0);
+ }
+ return new Pipeline(processors);
+ }
+
public Pipeline(Collection<Processor> processors) {
super(processors);
}
Modified: activemq/camel/trunk/camel-core/src/main/resources/org/apache/camel/model/jaxb.index
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/main/resources/org/apache/camel/model/jaxb.index?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/main/resources/org/apache/camel/model/jaxb.index (original)
+++ activemq/camel/trunk/camel-core/src/main/resources/org/apache/camel/model/jaxb.index Mon Jul 30 08:04:26 2007
@@ -4,6 +4,7 @@
FilterType
FromType
IdempotentConsumerType
+MulticastType
OtherwiseType
PolicyRef
ProcessorRef
Modified: activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/ErrorHandlerTest.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/ErrorHandlerTest.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/ErrorHandlerTest.java (original)
+++ activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/ErrorHandlerTest.java Mon Jul 30 08:04:26 2007
@@ -157,9 +157,12 @@
// START SNIPPET: e5
RouteBuilder builder = new RouteBuilder() {
public void configure() {
- inheritErrorHandler(false);
- from("seda:a").errorHandler(loggingErrorHandler("FOO.BAR")).filter(body().isInstanceOf(String.class)).to("seda:b");
+ from("seda:a").
+ errorHandler(loggingErrorHandler("FOO.BAR")).
+ filter(body().isInstanceOf(String.class)).
+ inheritErrorHandler(false).
+ to("seda:b");
}
};
// END SNIPPET: e5
Modified: activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/InterceptorBuilderTest.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/InterceptorBuilderTest.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/InterceptorBuilderTest.java (original)
+++ activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/InterceptorBuilderTest.java Mon Jul 30 08:04:26 2007
@@ -61,10 +61,16 @@
RouteBuilder builder = new RouteBuilder() {
public void configure() {
from("direct:a")
+ .intercept(interceptor1)
+ .intercept(interceptor2)
+ .to("direct:d");
+/*
+ TODO keep old DSL?
.intercept()
.add(interceptor1)
.add(interceptor2)
.target().to("direct:d");
+*/
}
};
container.addRoutes(builder);
Modified: activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/RouteBuilderTest.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/RouteBuilderTest.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/RouteBuilderTest.java (original)
+++ activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/builder/RouteBuilderTest.java Mon Jul 30 08:04:26 2007
@@ -237,10 +237,17 @@
RouteBuilder builder = new RouteBuilder() {
public void configure() {
from("seda:a")
+ .intercept(interceptor1)
+ .intercept(interceptor2)
+ .to("seda:d");
+/*
+
+ TODO keep old DSL?
.intercept()
.add(interceptor1)
.add(interceptor2)
.target().to("seda:d");
+*/
}
};
// END SNIPPET: e7
@@ -393,7 +400,7 @@
assertIsInstanceOf(MemoryMessageIdRepository.class, idempotentConsumer.getMessageIdRepository());
- SendProcessor sendProcessor = assertIsInstanceOf(SendProcessor.class, idempotentConsumer.getNextProcessor());
+ SendProcessor sendProcessor = assertIsInstanceOf(SendProcessor.class, unwrapErrorHandler(idempotentConsumer.getNextProcessor()));
assertEquals("Endpoint URI", "seda:b", sendProcessor.getDestination().getEndpointUri());
}
}
Modified: activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/MulticastTest.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/MulticastTest.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/MulticastTest.java (original)
+++ activemq/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/MulticastTest.java Mon Jul 30 08:04:26 2007
@@ -71,7 +71,7 @@
return new RouteBuilder() {
public void configure() {
// START SNIPPET: example
- from("direct:a").to("direct:x", "direct:y", "direct:z");
+ from("direct:a").multicast().to("direct:x", "direct:y", "direct:z");
// END SNIPPET: example
from("direct:x").process(processor).to("mock:x");
Modified: activemq/camel/trunk/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailMessage.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailMessage.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailMessage.java (original)
+++ activemq/camel/trunk/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailMessage.java Mon Jul 30 08:04:26 2007
@@ -56,6 +56,12 @@
return (MailExchange) super.getExchange();
}
+ public MailMessage copy() {
+ MailMessage answer = (MailMessage) super.copy();
+ answer.mailMessage = mailMessage;
+ return answer;
+ }
+
/**
* Returns the underlying Mail message
*
Modified: activemq/camel/trunk/components/camel-mail/src/test/java/org/apache/camel/component/mail/MultipleDestinationConsumeTest.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-mail/src/test/java/org/apache/camel/component/mail/MultipleDestinationConsumeTest.java?view=diff&rev=561006&r1=561005&r2=561006
==============================================================================
--- activemq/camel/trunk/components/camel-mail/src/test/java/org/apache/camel/component/mail/MultipleDestinationConsumeTest.java (original)
+++ activemq/camel/trunk/components/camel-mail/src/test/java/org/apache/camel/component/mail/MultipleDestinationConsumeTest.java Mon Jul 30 08:04:26 2007
@@ -70,6 +70,7 @@
MailExchange mailExchange = (MailExchange) exchange;
Message inMessage = mailExchange.getIn().getMessage();
+ assertNotNull("In message has no JavaMail message!", inMessage);
Enumeration iter = inMessage.getAllHeaders();
while (iter.hasMoreElements()) {
Header header = (Header) iter.nextElement();