Author: davsclaus
Date: Mon Jun 28 05:45:55 2010
New Revision: 958464
URL: http://svn.apache.org/viewvc?rev=958464&view=rev
Log:
CAMEL-2866: Filter and CBR EIPs support non blocking async routing engine.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointCBRTest.java
(with props)
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointFilterTest.java
- copied, changed from r958457,
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTryCatchFinallyTest.java
(with props)
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncCBRTest.java
(with props)
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncFilterTest.java
- copied, changed from r958457,
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ChoiceProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DelegateProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/FilterProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/TryProcessor.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ChoiceProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ChoiceProcessor.java?rev=958464&r1=958463&r2=958464&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ChoiceProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ChoiceProcessor.java
Mon Jun 28 05:45:55 2010
@@ -19,11 +19,15 @@ package org.apache.camel.processor;
import java.util.ArrayList;
import java.util.List;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
import org.apache.camel.Exchange;
import org.apache.camel.Navigate;
import org.apache.camel.Predicate;
import org.apache.camel.Processor;
import org.apache.camel.impl.ServiceSupport;
+import org.apache.camel.impl.converter.AsyncProcessorTypeConverter;
+import org.apache.camel.util.AsyncProcessorHelper;
import org.apache.camel.util.ServiceHelper;
/**
@@ -33,27 +37,33 @@ import org.apache.camel.util.ServiceHelp
*
* @version $Revision$
*/
-public class ChoiceProcessor extends ServiceSupport implements Processor,
Navigate<Processor>, Traceable {
+public class ChoiceProcessor extends ServiceSupport implements AsyncProcessor,
Navigate<Processor>, Traceable {
private final List<FilterProcessor> filters;
- private final Processor otherwise;
+ private final AsyncProcessor otherwise;
public ChoiceProcessor(List<FilterProcessor> filters, Processor otherwise)
{
this.filters = filters;
- this.otherwise = otherwise;
+ this.otherwise = AsyncProcessorTypeConverter.convert(otherwise);
}
public void process(Exchange exchange) throws Exception {
+ AsyncProcessorHelper.process(this, exchange);
+ }
+
+ public boolean process(Exchange exchange, AsyncCallback callback) {
for (FilterProcessor filterProcessor : filters) {
Predicate predicate = filterProcessor.getPredicate();
if (predicate != null && predicate.matches(exchange)) {
// process next will also take care (has not null test) if
next was a stop().
- // stop() has no processor to execute, and thus we will end in
a NPE
- filterProcessor.processNext(exchange);
- return;
+ // stop() has no processor to execute, and thus we will end in
a NPE
+ return filterProcessor.process(exchange, callback);
}
}
if (otherwise != null) {
- otherwise.process(exchange);
+ return otherwise.process(exchange, callback);
+ } else {
+ callback.done(true);
+ return true;
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DelegateProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DelegateProcessor.java?rev=958464&r1=958463&r2=958464&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DelegateProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/DelegateProcessor.java
Mon Jun 28 05:45:55 2010
@@ -28,6 +28,10 @@ import org.apache.camel.util.ServiceHelp
/**
* A Delegate pattern which delegates processing to a nested {...@link
Processor} which can
* be useful for implementation inheritance when writing an {...@link
org.apache.camel.spi.Policy}
+ * <p/>
+ * <b>Important:</b> This implementation does <b>not</b> support the
asynchronous routing engine.
+ * If you are implementing a EIP pattern please use the {...@link
org.apache.camel.processor.DelegateAsyncProcessor}
+ * instead.
*
* @version $Revision$
* @see org.apache.camel.processor.DelegateAsyncProcessor
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/FilterProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/FilterProcessor.java?rev=958464&r1=958463&r2=958464&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/FilterProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/FilterProcessor.java
Mon Jun 28 05:45:55 2010
@@ -16,6 +16,7 @@
*/
package org.apache.camel.processor;
+import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.Predicate;
import org.apache.camel.Processor;
@@ -29,7 +30,7 @@ import org.apache.commons.logging.LogFac
*
* @version $Revision$
*/
-public class FilterProcessor extends DelegateProcessor implements Traceable {
+public class FilterProcessor extends DelegateAsyncProcessor implements
Traceable {
private static final Log LOG = LogFactory.getLog(FilterProcessor.class);
private final Predicate predicate;
@@ -38,7 +39,8 @@ public class FilterProcessor extends Del
this.predicate = predicate;
}
- public void process(Exchange exchange) throws Exception {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
boolean matches = predicate.matches(exchange);
if (LOG.isDebugEnabled()) {
@@ -46,7 +48,10 @@ public class FilterProcessor extends Del
}
if (matches) {
- super.process(exchange);
+ return super.process(exchange, callback);
+ } else {
+ callback.done(true);
+ return true;
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java?rev=958464&r1=958463&r2=958464&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
Mon Jun 28 05:45:55 2010
@@ -29,13 +29,8 @@ import org.apache.camel.util.AsyncProces
import org.apache.camel.util.ObjectHelper;
/**
- * Threads processor that leverage a thread pool for processing exchanges.
- * <p/>
- * The original caller thread will receive a <tt>Future<Exchange></tt>
in the OUT message body.
- * It can then later use this handle to obtain the async response.
- * <p/>
- * Camel also provides type converters so you can just ask to get the desired
object type and Camel
- * will automatic wait for the async task to complete to return the response.
+ * Threads processor that leverage a thread pool for continue processing the
{...@link Exchange}s
+ * using the asynchronous routing engine.
*
* @version $Revision$
*/
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/TryProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/TryProcessor.java?rev=958464&r1=958463&r2=958464&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/TryProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/TryProcessor.java
Mon Jun 28 05:45:55 2010
@@ -36,6 +36,8 @@ import org.apache.commons.logging.LogFac
public class TryProcessor extends ServiceSupport implements Processor,
Navigate<Processor>, Traceable {
private static final transient Log LOG =
LogFactory.getLog(TryProcessor.class);
+ // TODO: support async routing engine
+
protected final Processor tryProcessor;
protected final List<CatchProcessor> catchClauses;
protected final Processor finallyProcessor;
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointCBRTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointCBRTest.java?rev=958464&view=auto
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointCBRTest.java
(added)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointCBRTest.java
Mon Jun 28 05:45:55 2010
@@ -0,0 +1,76 @@
+/**
+ * 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.async;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+
+/**
+ * @version $Revision$
+ */
+public class AsyncEndpointCBRTest extends ContextTestSupport {
+
+ private static String beforeThreadName;
+ private static String afterThreadName;
+
+ public void testAsyncEndpoint() throws Exception {
+ getMockEndpoint("mock:before").expectedBodiesReceived("Hello Camel");
+ getMockEndpoint("mock:after").expectedBodiesReceived("Bye Camel");
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye Camel");
+
+ String reply = template.requestBody("direct:start", "Hello Camel",
String.class);
+ assertEquals("Bye Camel", reply);
+
+ assertMockEndpointsSatisfied();
+
+ assertFalse("Should use different threads",
beforeThreadName.equalsIgnoreCase(afterThreadName));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ from("direct:start")
+ .to("mock:before")
+ .to("log:before")
+ .choice()
+ .when(body().contains("Camel"))
+ .process(new Processor() {
+ public void process(Exchange exchange)
throws Exception {
+ beforeThreadName =
Thread.currentThread().getName();
+ }
+ })
+ .to("async:Bye Camel")
+ .process(new Processor() {
+ public void process(Exchange exchange)
throws Exception {
+ afterThreadName =
Thread.currentThread().getName();
+ }
+ })
+ .to("log:after")
+ .to("mock:after")
+ .end()
+ .to("mock:result");
+ }
+ };
+ }
+
+}
\ No newline at end of file
Propchange:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointCBRTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointCBRTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Copied:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointFilterTest.java
(from r958457,
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTest.java)
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointFilterTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointFilterTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTest.java&r1=958457&r2=958464&rev=958464&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointFilterTest.java
Mon Jun 28 05:45:55 2010
@@ -24,7 +24,7 @@ import org.apache.camel.builder.RouteBui
/**
* @version $Revision$
*/
-public class AsyncEndpointTest extends ContextTestSupport {
+public class AsyncEndpointFilterTest extends ContextTestSupport {
private static String beforeThreadName;
private static String afterThreadName;
@@ -52,22 +52,24 @@ public class AsyncEndpointTest extends C
from("direct:start")
.to("mock:before")
.to("log:before")
- .process(new Processor() {
- public void process(Exchange exchange) throws
Exception {
- beforeThreadName =
Thread.currentThread().getName();
- }
- })
- .to("async:Bye Camel")
- .process(new Processor() {
- public void process(Exchange exchange) throws
Exception {
- afterThreadName =
Thread.currentThread().getName();
- }
- })
- .to("log:after")
- .to("mock:after")
+ .filter(body().contains("Camel"))
+ .process(new Processor() {
+ public void process(Exchange exchange) throws
Exception {
+ beforeThreadName =
Thread.currentThread().getName();
+ }
+ })
+ .to("async:Bye Camel")
+ .process(new Processor() {
+ public void process(Exchange exchange) throws
Exception {
+ afterThreadName =
Thread.currentThread().getName();
+ }
+ })
+ .to("log:after")
+ .to("mock:after")
+ .end()
.to("mock:result");
}
};
}
-}
+}
\ No newline at end of file
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTryCatchFinallyTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTryCatchFinallyTest.java?rev=958464&view=auto
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTryCatchFinallyTest.java
(added)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTryCatchFinallyTest.java
Mon Jun 28 05:45:55 2010
@@ -0,0 +1,79 @@
+/**
+ * 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.async;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+
+/**
+ * @version $Revision$
+ */
+public class AsyncEndpointTryCatchFinallyTest extends ContextTestSupport {
+
+ private static String beforeThreadName;
+ private static String afterThreadName;
+
+ public void testAsyncEndpoint() throws Exception {
+ // TODO: try catch finally need to support async routing engine
+
+ getMockEndpoint("mock:before").expectedBodiesReceived("Hello Camel");
+ //getMockEndpoint("mock:after").expectedBodiesReceived("Bye Camel");
+ //getMockEndpoint("mock:result").expectedBodiesReceived("Bye Camel");
+
+ String reply = template.requestBody("direct:start", "Hello Camel",
String.class);
+ //assertEquals("Bye Camel", reply);
+
+ assertMockEndpointsSatisfied();
+
+ //assertFalse("Should use different threads",
beforeThreadName.equalsIgnoreCase(afterThreadName));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ from("direct:start")
+ .to("mock:before")
+ .to("log:before")
+ .doTry()
+ .process(new Processor() {
+ public void process(Exchange exchange) throws
Exception {
+ beforeThreadName =
Thread.currentThread().getName();
+ }
+ })
+ .to("async:Bye Camel?failFirstAttempts=1")
+ .doCatch(Exception.class)
+ .process(new Processor() {
+ public void process(Exchange exchange) throws
Exception {
+ afterThreadName =
Thread.currentThread().getName();
+ }
+ })
+ .doFinally()
+ .to("log:after")
+ .to("mock:after")
+ .end()
+ .to("mock:result");
+ }
+ };
+ }
+
+}
\ No newline at end of file
Propchange:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTryCatchFinallyTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/AsyncEndpointTryCatchFinallyTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added:
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncCBRTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncCBRTest.java?rev=958464&view=auto
==============================================================================
---
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncCBRTest.java
(added)
+++
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncCBRTest.java
Mon Jun 28 05:45:55 2010
@@ -0,0 +1,55 @@
+/**
+ * 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.component.jetty.async;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+/**
+ * @version $Revision$
+ */
+public class JettyAsyncCBRTest extends CamelTestSupport {
+
+ @Test
+ public void testJettyAsync() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye World");
+
+ String reply = template.requestBody("http://localhost:8876/myservice";,
"Hello Camel", String.class);
+ assertEquals("Bye World", reply);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addComponent("async", new MyAsyncComponent());
+
+ from("jetty:http://localhost:8876/myservice";)
+ .convertBodyTo(String.class)
+ .choice()
+ .when(body().contains("Camel"))
+ .to("async:Bye World")
+ .end()
+ .to("mock:result");
+ }
+ };
+ }
+}
\ No newline at end of file
Propchange:
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncCBRTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncCBRTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Copied:
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncFilterTest.java
(from r958457,
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncTest.java)
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncFilterTest.java?p2=camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncFilterTest.java&p1=camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncTest.java&r1=958457&r2=958464&rev=958464&view=diff
==============================================================================
---
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncTest.java
(original)
+++
camel/trunk/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/async/JettyAsyncFilterTest.java
Mon Jun 28 05:45:55 2010
@@ -23,13 +23,13 @@ import org.junit.Test;
/**
* @version $Revision$
*/
-public class JettyAsyncTest extends CamelTestSupport {
+public class JettyAsyncFilterTest extends CamelTestSupport {
@Test
public void testJettyAsync() throws Exception {
getMockEndpoint("mock:result").expectedBodiesReceived("Bye World");
- String reply = template.requestBody("http://localhost:8877/myservice";,
null, String.class);
+ String reply = template.requestBody("http://localhost:8876/myservice";,
"Hello Camel", String.class);
assertEquals("Bye World", reply);
assertMockEndpointsSatisfied();
@@ -42,10 +42,13 @@ public class JettyAsyncTest extends Came
public void configure() throws Exception {
context.addComponent("async", new MyAsyncComponent());
- from("jetty:http://localhost:8877/myservice";)
- .to("async:Bye World")
+ from("jetty:http://localhost:8876/myservice";)
+ .convertBodyTo(String.class)
+ .filter(body().contains("Camel"))
+ .to("async:Bye World")
+ .end()
.to("mock:result");
}
};
}
-}
+}
\ No newline at end of file
