Author: davsclaus
Date: Wed Aug 5 06:06:58 2009
New Revision: 801070
URL: http://svn.apache.org/viewvc?rev=801070&view=rev
Log:
CAMEL-1876: Added asyncCallback methods to ProducerTemplate to do async
messaging with callbacks being invoked when the message is done. This alows
client to not hazzle with how to wait for the response as the callback is
invoked when the message is done.
Added:
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/async/HttpAsyncCallbackTest.java
(with props)
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java?rev=801070&r1=801069&r2=801070&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultProducerTemplate.java
Wed Aug 5 06:06:58 2009
@@ -398,16 +398,6 @@
};
}
- protected Processor createSetBodyProcessorCallback(final Object body,
final Synchronization onCompletion) {
- return new Processor() {
- public void process(Exchange exchange) {
- Message in = exchange.getIn();
- in.setBody(body);
- exchange.addOnCompletion(onCompletion);
- }
- };
- }
-
protected Endpoint resolveMandatoryEndpoint(String endpointUri) {
Endpoint endpoint = context.getEndpoint(endpointUri);
if (endpoint == null) {
@@ -613,9 +603,19 @@
public Future<Object> asyncCallback(final Endpoint endpoint, final
ExchangePattern pattern, final Object body, final Synchronization onCompletion)
{
Callable<Object> task = new Callable<Object>() {
public Object call() throws Exception {
- Exchange exchange = send(endpoint, pattern,
createSetBodyProcessorCallback(body, onCompletion));
+ Exchange answer = send(endpoint, pattern,
createSetBodyProcessor(body));
+
+ // invoke callback before returning answer
+ // as it allows callback to be used without
UnitOfWorkProcessor invoking it
+ // and thus it works directly from a producer template as
well, as opposed
+ // to the UnitOfWorkProcessor that is injected in routes
+ if (answer.isFailed()) {
+ onCompletion.onFailure(answer);
+ } else {
+ onCompletion.onComplete(answer);
+ }
- Object result = extractResultBody(exchange, pattern);
+ Object result = extractResultBody(answer, pattern);
if (pattern.isOutCapable()) {
return result;
} else {
@@ -631,8 +631,18 @@
public Future<Exchange> asyncCallback(final Endpoint endpoint, final
Exchange exchange, final Synchronization onCompletion) {
Callable<Exchange> task = new Callable<Exchange>() {
public Exchange call() throws Exception {
- exchange.addOnCompletion(onCompletion);
- return send(endpoint, exchange);
+ Exchange answer = send(endpoint, exchange);
+
+ // invoke callback before returning answer
+ // as it allows callback to be used without
UnitOfWorkProcessor invoking it
+ // and thus it works directly from a producer template as
well, as opposed
+ // to the UnitOfWorkProcessor that is injected in routes
+ if (answer.isFailed()) {
+ onCompletion.onFailure(answer);
+ } else {
+ onCompletion.onComplete(answer);
+ }
+ return answer;
}
};
@@ -642,7 +652,18 @@
public Future<Exchange> asyncCallback(final Endpoint endpoint, final
Processor processor, final Synchronization onCompletion) {
Callable<Exchange> task = new Callable<Exchange>() {
public Exchange call() throws Exception {
- return producerCache.send(endpoint, null, processor,
onCompletion);
+ Exchange answer = producerCache.send(endpoint, processor);
+
+ // invoke callback before returning answer
+ // as it allows callback to be used without
UnitOfWorkProcessor invoking it
+ // and thus it works directly from a producer template as
well, as opposed
+ // to the UnitOfWorkProcessor that is injected in routes
+ if (answer.isFailed()) {
+ onCompletion.onFailure(answer);
+ } else {
+ onCompletion.onComplete(answer);
+ }
+ return answer;
}
};
Added:
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/async/HttpAsyncCallbackTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/async/HttpAsyncCallbackTest.java?rev=801070&view=auto
==============================================================================
---
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/async/HttpAsyncCallbackTest.java
(added)
+++
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/async/HttpAsyncCallbackTest.java
Wed Aug 5 06:06:58 2009
@@ -0,0 +1,100 @@
+/**
+ * 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.itest.async;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.SynchronizationAdapter;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+/**
+ * @version $Revision$
+ */
+public class HttpAsyncCallbackTest extends CamelTestSupport {
+
+ @Test
+ public void testAsyncAndSyncAtSameTimeWithHttp() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("Hello Claus", "Hello Hadrian",
"Hello Willem");
+
+ // START SNIPPET: e3
+ MyCallback callback = new MyCallback();
+
+ // Send 3 async request/reply message to the http endpoint
+ // where we let the callback handle gathering the responses
+ template.asyncCallbackRequestBody("http://localhost:9080/myservice";,
"Claus", callback);
+ template.asyncCallbackRequestBody("http://localhost:9080/myservice";,
"Hadrian", callback);
+ template.asyncCallbackRequestBody("http://localhost:9080/myservice";,
"Willem", callback);
+
+ // END SNIPPET: e3
+ assertMockEndpointsSatisfied();
+
+ // give on completion time to complete properly before we do
assertions on its size
+ Thread.sleep(1000);
+
+ // assert that we got all the correct data in our callback
+ assertEquals(3, callback.getData().size());
+ assertTrue("Claus is missing", callback.getData().contains("Hello
Claus"));
+ assertTrue("Hadrian is missing", callback.getData().contains("Hello
Hadrian"));
+ assertTrue("Willem is missing", callback.getData().contains("Hello
Willem"));
+ }
+
+ // START SNIPPET: e2
+ /**
+ * Our own callback that will gather all the responses.
+ * We extend the SynchronizationAdapter class as we then only need to
override the onComplete method.
+ */
+ private class MyCallback extends SynchronizationAdapter {
+
+ private List<String> data = new ArrayList<String>();
+
+ @Override
+ public void onDone(Exchange exchange) {
+ // this method is invoked when the exchange was a success and we
can get the response
+ String body = exchange.getOut().getBody(String.class);
+ data.add(body);
+ }
+
+ public List<String> getData() {
+ return data;
+ }
+ }
+ // END SNIPPET: e2
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ // START SNIPPET: e1
+ // The mocks are here for unit test
+ // Simulate a slow http service (delaying 1 sec) we want to
invoke async
+ from("jetty:http://0.0.0.0:9080/myservice";)
+ .delay(1000)
+ .transform(body().prepend("Hello "))
+ .to("mock:result");
+ // END SNIPPET: e1
+ }
+ };
+ }
+
+}
Propchange:
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/async/HttpAsyncCallbackTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
camel/trunk/tests/camel-itest/src/test/java/org/apache/camel/itest/async/HttpAsyncCallbackTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
