camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [1/2] camel git commit: CAMEL-10286 Allow async bean method in bean language with J8 CompletableFuture
Date Mon, 05 Sep 2016 07:36:58 GMT
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 <vit@tym.im>
Authored: Sun Sep 4 15:02:47 2016 -0400
Committer: Claus Ibsen <davsclaus@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


Mime
View raw message