You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2016/09/05 07:36:58 UTC
[1/2] camel git commit: CAMEL-10286 Allow async bean method in bean
language with J8 CompletableFuture
Repository: camel
Updated Branches:
refs/heads/master 6e1c34413 -> ab0aeca64
CAMEL-10286
Allow async bean method in bean language with J8 CompletableFuture
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/41634f2e
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/41634f2e
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/41634f2e
Branch: refs/heads/master
Commit: 41634f2ee5e1736398a1e4bc1678699483901b7f
Parents: 6e1c344
Author: Vitalii Tymchyshyn <vi...@tym.im>
Authored: Sun Sep 4 15:02:47 2016 -0400
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Sep 5 09:28:24 2016 +0200
----------------------------------------------------------------------
.../apache/camel/component/bean/MethodInfo.java | 43 ++++--
.../component/bean/BeanInvokeAsyncTest.java | 135 +++++++++++++++++++
2 files changed, 167 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/41634f2e/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java b/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
index 368323a..d4a0c57 100644
--- a/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
+++ b/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
@@ -28,6 +28,7 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutorService;
import org.apache.camel.AsyncCallback;
@@ -298,19 +299,25 @@ public class MethodInfo {
return routingSlip.doRoutingSlip(exchange, result, callback);
}
+ //If it's Java 8 async result
+ if (CompletionStage.class.isAssignableFrom(getMethod().getReturnType())) {
+ CompletionStage<?> completionStage = (CompletionStage<?>) result;
+
+ completionStage
+ .whenComplete((resultObject, e) -> {
+ if (e != null) {
+ exchange.setException(e);
+ } else if (resultObject != null){
+ fillResult(exchange, resultObject);
+ }
+ callback.done(false);
+ });
+ return false;
+ }
+
// if the method returns something then set the value returned on the Exchange
if (!getMethod().getReturnType().equals(Void.TYPE) && result != Void.TYPE) {
- if (exchange.getPattern().isOutCapable()) {
- // force out creating if not already created (as its lazy)
- LOG.debug("Setting bean invocation result on the OUT message: {}", result);
- exchange.getOut().setBody(result);
- // propagate headers
- exchange.getOut().getHeaders().putAll(exchange.getIn().getHeaders());
- } else {
- // if not out then set it on the in
- LOG.debug("Setting bean invocation result on the IN message: {}", result);
- exchange.getIn().setBody(result);
- }
+ fillResult(exchange, result);
}
// we did not use any of the eips, but just invoked the bean
@@ -329,6 +336,20 @@ public class MethodInfo {
};
}
+ private void fillResult(Exchange exchange, Object result) {
+ if (exchange.getPattern().isOutCapable()) {
+ // force out creating if not already created (as its lazy)
+ LOG.debug("Setting bean invocation result on the OUT message: {}", result);
+ exchange.getOut().setBody(result);
+ // propagate headers
+ exchange.getOut().getHeaders().putAll(exchange.getIn().getHeaders());
+ } else {
+ // if not out then set it on the in
+ LOG.debug("Setting bean invocation result on the IN message: {}", result);
+ exchange.getIn().setBody(result);
+ }
+ }
+
public Class<?> getType() {
return type;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/41634f2e/camel-core/src/test/java/org/apache/camel/component/bean/BeanInvokeAsyncTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/component/bean/BeanInvokeAsyncTest.java b/camel-core/src/test/java/org/apache/camel/component/bean/BeanInvokeAsyncTest.java
new file mode 100644
index 0000000..a8ecf48
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/component/bean/BeanInvokeAsyncTest.java
@@ -0,0 +1,135 @@
+/**
+ * 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.bean;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.Assert;
+
+import java.util.concurrent.*;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+/**
+ * @version
+ */
+public class BeanInvokeAsyncTest extends ContextTestSupport {
+
+ private volatile CompletableFuture<Object> callFuture;
+ private volatile String receivedBody;
+ private volatile CountDownLatch methodInvoked;
+ private Future<Object> sendFuture;
+
+ public void testDoSomething() throws Exception {
+
+ runTestSendBody("Hello World", "Hello World", this::doSomething);
+ runTestSendBody("", "", this::doSomething);
+ runTestSendBody(this::expectNullBody, null, this::doSomething);
+ }
+
+ public void testChangeSomething() throws Exception {
+ runTestSendBody("Bye World", "Hello World", this::changeSomething);
+ runTestSendBody("Bye All", null, this::changeSomething);
+ runTestSendBody("Bye All", "", this::changeSomething);
+
+ }
+
+ public void testDoNothing() throws Exception {
+ runTestSendBody("Hello World", "Hello World", this::doNothing);
+ runTestSendBody("", "", this::doNothing);
+ runTestSendBody(this::expectNullBody, null, this::doNothing);
+ }
+
+ public void testThrowSomething() throws Exception {
+ try {
+ runTestSendBody(m -> m.expectedMessageCount(0), "SomeProblem", this::throwSomething);
+ fail("Exception expected");
+ } catch (ExecutionException e) {
+ Assert.assertTrue(e.getCause() instanceof CamelExecutionException);
+ Assert.assertTrue(e.getCause().getCause() instanceof IllegalStateException);
+ Assert.assertEquals("SomeProblem", e.getCause().getCause().getMessage());
+ }
+ }
+
+ private void runTestSendBody(String expectedBody, String sentBody, Function<String, String> processor)
+ throws InterruptedException, java.util.concurrent.ExecutionException {
+ runTestSendBody(m -> m.expectedBodiesReceived(expectedBody), sentBody, processor);
+ }
+
+ private void runTestSendBody(Consumer<MockEndpoint> mockPreparer, String sentBody,
+ Function<String, String> processor)
+ throws InterruptedException, java.util.concurrent.ExecutionException {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.reset();
+ mockPreparer.accept(mock);
+ callFuture = new CompletableFuture<>();
+ methodInvoked = new CountDownLatch(1);
+ sendFuture = template.asyncSendBody("direct:entry", sentBody);
+ Assert.assertTrue(methodInvoked.await(5, TimeUnit.SECONDS));
+ Assert.assertEquals(0, mock.getReceivedCounter());
+ Assert.assertFalse(sendFuture.isDone());
+ try {
+ callFuture.complete(processor.apply(receivedBody));
+ } catch (Exception e) {
+ callFuture.completeExceptionally(e);
+ }
+ sendFuture.get();
+ assertMockEndpointsSatisfied();
+ }
+
+ private void expectNullBody(MockEndpoint mock) {
+ mock.expectedMessageCount(1);
+ mock.message(0).body().isNull();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:entry").bean(BeanInvokeAsyncTest.this, "asyncMethod").to("mock:result");
+ }
+ };
+ }
+
+ public CompletableFuture<?> asyncMethod(String body) {
+ this.receivedBody = body;
+ methodInvoked.countDown();
+ return callFuture;
+ }
+
+ public String doSomething(String s) {
+ return s;
+ }
+
+ public String changeSomething(String s) {
+ if ("Hello World".equals(s)) {
+ return "Bye World";
+ }
+ return "Bye All";
+ }
+
+ public String doNothing(String s) {
+ return null;
+ }
+
+ public String throwSomething(String s) {
+ throw new IllegalStateException(s);
+ }
+}
\ No newline at end of file
[2/2] camel git commit: Fixed CS. This closes #1154
Posted by da...@apache.org.
Fixed CS. This closes #1154
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/ab0aeca6
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/ab0aeca6
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/ab0aeca6
Branch: refs/heads/master
Commit: ab0aeca6432cb41d8c7086e31a4cd3cba1288d46
Parents: 41634f2
Author: Claus Ibsen <da...@apache.org>
Authored: Mon Sep 5 09:35:44 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Mon Sep 5 09:35:44 2016 +0200
----------------------------------------------------------------------
.../apache/camel/component/bean/MethodInfo.java | 2 +-
.../component/bean/BeanInvokeAsyncTest.java | 25 +++++++++++++-------
2 files changed, 17 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/ab0aeca6/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java b/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
index d4a0c57..532046b 100644
--- a/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
+++ b/camel-core/src/main/java/org/apache/camel/component/bean/MethodInfo.java
@@ -307,7 +307,7 @@ public class MethodInfo {
.whenComplete((resultObject, e) -> {
if (e != null) {
exchange.setException(e);
- } else if (resultObject != null){
+ } else if (resultObject != null) {
fillResult(exchange, resultObject);
}
callback.done(false);
http://git-wip-us.apache.org/repos/asf/camel/blob/ab0aeca6/camel-core/src/test/java/org/apache/camel/component/bean/BeanInvokeAsyncTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/component/bean/BeanInvokeAsyncTest.java b/camel-core/src/test/java/org/apache/camel/component/bean/BeanInvokeAsyncTest.java
index a8ecf48..1ce151c 100644
--- a/camel-core/src/test/java/org/apache/camel/component/bean/BeanInvokeAsyncTest.java
+++ b/camel-core/src/test/java/org/apache/camel/component/bean/BeanInvokeAsyncTest.java
@@ -16,18 +16,22 @@
*/
package org.apache.camel.component.bean;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
import org.apache.camel.CamelExecutionException;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.junit.Assert;
-import java.util.concurrent.*;
-import java.util.function.Consumer;
-import java.util.function.Function;
-
/**
- * @version
+ * Unit test for Java 8 {@link CompletableFuture} as return type on a bean being called from a Camel route.
*/
public class BeanInvokeAsyncTest extends ContextTestSupport {
@@ -67,20 +71,21 @@ public class BeanInvokeAsyncTest extends ContextTestSupport {
}
}
- private void runTestSendBody(String expectedBody, String sentBody, Function<String, String> processor)
- throws InterruptedException, java.util.concurrent.ExecutionException {
+ private void runTestSendBody(String expectedBody, String sentBody,
+ Function<String, String> processor) throws Exception {
runTestSendBody(m -> m.expectedBodiesReceived(expectedBody), sentBody, processor);
}
private void runTestSendBody(Consumer<MockEndpoint> mockPreparer, String sentBody,
- Function<String, String> processor)
- throws InterruptedException, java.util.concurrent.ExecutionException {
+ Function<String, String> processor) throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
mock.reset();
mockPreparer.accept(mock);
+
callFuture = new CompletableFuture<>();
methodInvoked = new CountDownLatch(1);
sendFuture = template.asyncSendBody("direct:entry", sentBody);
+
Assert.assertTrue(methodInvoked.await(5, TimeUnit.SECONDS));
Assert.assertEquals(0, mock.getReceivedCounter());
Assert.assertFalse(sendFuture.isDone());
@@ -90,6 +95,7 @@ public class BeanInvokeAsyncTest extends ContextTestSupport {
callFuture.completeExceptionally(e);
}
sendFuture.get();
+
assertMockEndpointsSatisfied();
}
@@ -108,6 +114,7 @@ public class BeanInvokeAsyncTest extends ContextTestSupport {
};
}
+ // java 8 async return type
public CompletableFuture<?> asyncMethod(String body) {
this.receivedBody = body;
methodInvoked.countDown();