Author: davsclaus
Date: Tue Nov 10 11:13:06 2009
New Revision: 834430
URL: http://svn.apache.org/viewvc?rev=834430&view=rev
Log:
CAMEL-2151: Using Camel type converter to get the async processor.
Added:
camel/trunk/camel-core/src/main/java/org/apache/camel/converter/AsyncProcessorConverter.java
(with props)
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncHandoverCompletionTest.java
- copied, changed from r834396,
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncTest.java
Added:
camel/trunk/camel-core/src/main/java/org/apache/camel/converter/AsyncProcessorConverter.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/converter/AsyncProcessorConverter.java?rev=834430&view=auto
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/converter/AsyncProcessorConverter.java
(added)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/converter/AsyncProcessorConverter.java
Tue Nov 10 11:13:06 2009
@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.converter;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.Converter;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.processor.DelegateProcessor;
+
+/**
+ * @version $Revision$
+ */
+...@converter
+public class AsyncProcessorConverter {
+
+ private static final class ProcessorToAsyncProcessorBridge extends
DelegateProcessor implements AsyncProcessor {
+
+ private ProcessorToAsyncProcessorBridge(Processor processor) {
+ super(processor);
+ }
+
+ public void process(Exchange exchange, AsyncCallback callback) throws
Exception {
+ try {
+ processor.process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
+ callback.onDataReceived(exchange);
+ }
+ }
+
+ @Converter
+ public static AsyncProcessor toAsyncProcessor(final Processor processor) {
+ if (processor instanceof AsyncProcessor) {
+ return (AsyncProcessor) processor;
+ } else {
+ return new ProcessorToAsyncProcessorBridge(processor);
+ }
+ }
+
+}
Propchange:
camel/trunk/camel-core/src/main/java/org/apache/camel/converter/AsyncProcessorConverter.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
camel/trunk/camel-core/src/main/java/org/apache/camel/converter/AsyncProcessorConverter.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java?rev=834430&r1=834429&r2=834430&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
Tue Nov 10 11:13:06 2009
@@ -84,24 +84,19 @@
public Exchange doInProducer(Producer producer, Exchange exchange,
ExchangePattern pattern) throws Exception {
exchange = configureExchange(exchange, pattern);
- if (producer instanceof AsyncProcessor) {
- // let the producer use this callback to signal completion
- AsyncProcessor asyncProcessor = (AsyncProcessor) producer;
-
- // pass in the callback that adds the exchange to the
completed list of tasks
- final AsyncCallback callback = new AsyncCallback() {
- public void onDataReceived(Exchange exchange) {
- completedTasks.add(exchange);
- }
- };
-
- asyncProcessor.process(exchange, callback);
- } else {
- // its not a real AsyncProcessor so simulate async
processing
- producer.process(exchange);
- completedTasks.add(exchange);
- }
+ AsyncProcessor asyncProducer =
exchange.getContext().getTypeConverter().convertTo(AsyncProcessor.class,
producer);
+
+ // pass in the callback that adds the exchange to the
completed list of tasks
+ final AsyncCallback callback = new AsyncCallback() {
+ public void onDataReceived(Exchange exchange) {
+ completedTasks.add(exchange);
+ }
+ };
+
+ // produce it async
+ asyncProducer.process(exchange, callback);
+ // and return the exchange
return exchange;
}
});
Copied:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncHandoverCompletionTest.java
(from r834396,
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncTest.java)
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncHandoverCompletionTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncHandoverCompletionTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncTest.java&r1=834396&r2=834430&rev=834430&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncHandoverCompletionTest.java
Tue Nov 10 11:13:06 2009
@@ -16,34 +16,36 @@
*/
package org.apache.camel.processor.async;
+import java.io.File;
+
import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+
+import static org.apache.camel.language.simple.SimpleLanguage.simple;
/**
* @version $Revision$
*/
-public class ToAsyncTest extends ContextTestSupport {
+public class ToAsyncHandoverCompletionTest extends ContextTestSupport {
- public void testToAsync() throws Exception {
- getMockEndpoint("mock:a").expectedBodiesReceived("Hello World");
- getMockEndpoint("mock:b").expectedBodiesReceived("Hello World");
- getMockEndpoint("mock:result").expectedMessageCount(1);
-
getMockEndpoint("mock:result").message(0).outBody(String.class).isEqualTo("Bye
World");
+ public void testToAsyncHandoverCompletion() throws Exception {
+ deleteDirectory("target/toasync");
- template.sendBody("direct:start", "Hello World");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("Bye World");
+ mock.expectedFileExists("target/toasync/done/hello.txt");
- assertMockEndpointsSatisfied();
+ template.sendBodyAndHeader("file://target/toasync", "World",
Exchange.FILE_NAME, "hello.txt");
- // and it should be different exchange ids
+ Thread.sleep(1000);
- String ida =
getMockEndpoint("mock:a").getReceivedExchanges().get(0).getExchangeId();
- String idb =
getMockEndpoint("mock:b").getReceivedExchanges().get(0).getExchangeId();
- String idresult =
getMockEndpoint("mock:result").getReceivedExchanges().get(0).getExchangeId();
-
- // id a should be different and id b and id result the same
- assertNotSame(ida, idb);
- assertNotSame(ida, idresult);
- assertSame(idb, idresult);
+ // now there is a delay of 3 seconds but the original file should
still be there as its in progress
+ File target = new File("target/toasync/hello.txt").getAbsoluteFile();
+ assertEquals(true, target.exists());
+
+ assertMockEndpointsSatisfied();
}
@Override
@@ -51,10 +53,10 @@
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:start").to("mock:a").toAsync("direct:bar",
5).to("mock:result");
+
from("file://target/toasync?move=done").to("mock:a").toAsync("direct:bar",
5).to("mock:result");
- from("direct:bar").to("mock:b").transform(constant("Bye
World"));
+ from("direct:bar").delay(3000).transform(simple("Bye
${body}"));
}
};
}
-}
+}
\ No newline at end of file
Modified:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncTest.java?rev=834430&r1=834429&r2=834430&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/async/ToAsyncTest.java
Tue Nov 10 11:13:06 2009
@@ -27,8 +27,7 @@
public void testToAsync() throws Exception {
getMockEndpoint("mock:a").expectedBodiesReceived("Hello World");
getMockEndpoint("mock:b").expectedBodiesReceived("Hello World");
- getMockEndpoint("mock:result").expectedMessageCount(1);
-
getMockEndpoint("mock:result").message(0).outBody(String.class).isEqualTo("Bye
World");
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye World");
template.sendBody("direct:start", "Hello World");
