camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [2/7] camel git commit: Camel experiment with using vertx as eventbus for routing engine.
Date Wed, 13 Jul 2016 12:46:51 GMT
Camel experiment with using vertx as eventbus for routing engine.


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/3e3e1803
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/3e3e1803
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/3e3e1803

Branch: refs/heads/eventbus
Commit: 3e3e1803316e45035d9a51c101a1eb38d19c8508
Parents: 48219df
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Fri Apr 8 10:20:09 2016 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Fri Apr 8 10:20:09 2016 +0200

----------------------------------------------------------------------
 .../vertx/eventbus/VertxCamelProducer.java      |   6 ++
 .../vertx/eventbus/VertxCamelTransform.java     | 102 +++++++++++++++++++
 .../vertx/eventbus/VertxProcessorFactory.java   |   5 +
 .../vertx/eventbus/VertxSendToProcessor.java    |  22 +++-
 .../vertx/eventbus/VertxTransformProcessor.java |  60 +++++++++++
 .../vertx/eventbus/VertxTransformTest.java      |  71 +++++++++++++
 6 files changed, 262 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/3e3e1803/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxCamelProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxCamelProducer.java
b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxCamelProducer.java
index 923ca38..7b411c9 100644
--- a/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxCamelProducer.java
+++ b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxCamelProducer.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.vertx.eventbus;
 
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
+import io.vertx.core.eventbus.DeliveryOptions;
 import io.vertx.core.eventbus.Message;
 import io.vertx.core.eventbus.MessageConsumer;
 import org.apache.camel.CamelContext;
@@ -33,12 +34,15 @@ public class VertxCamelProducer extends ServiceSupport implements Handler<Messag
     private final Vertx vertx;
     private final String id;
     private MessageConsumer<Exchange> consumer;
+    private final DeliveryOptions options;
 
     public VertxCamelProducer(CamelContext camelContext, Vertx vertx, String id) {
         this.camelContext = camelContext;
         this.template = camelContext.createProducerTemplate();
         this.vertx = vertx;
         this.id = id;
+        this.options = new DeliveryOptions();
+        this.options.setCodecName("camel");
     }
 
     @Override
@@ -61,5 +65,7 @@ public class VertxCamelProducer extends ServiceSupport implements Handler<Messag
         String url = (String) exchange.removeProperty("CamelVertxUrl");
         // TODO: execute blocking
         template.send(url, exchange);
+        // signal we are done
+        event.reply(exchange, options);
     }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/3e3e1803/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxCamelTransform.java
----------------------------------------------------------------------
diff --git a/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxCamelTransform.java
b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxCamelTransform.java
new file mode 100644
index 0000000..923fcaf
--- /dev/null
+++ b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxCamelTransform.java
@@ -0,0 +1,102 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.vertx.eventbus;
+
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
+import io.vertx.core.eventbus.DeliveryOptions;
+import io.vertx.core.eventbus.Message;
+import io.vertx.core.eventbus.MessageConsumer;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.impl.DefaultMessage;
+import org.apache.camel.support.ServiceSupport;
+import org.apache.camel.util.ExchangeHelper;
+
+public class VertxCamelTransform extends ServiceSupport implements Handler<Message<Exchange>>
{
+
+    private final CamelContext camelContext;
+    private final Vertx vertx;
+    private final String id;
+    private MessageConsumer<Exchange> consumer;
+    private final DeliveryOptions options;
+
+    public VertxCamelTransform(CamelContext camelContext, Vertx vertx, String id) {
+        this.camelContext = camelContext;
+        this.vertx = vertx;
+        this.id = id;
+        this.options = new DeliveryOptions();
+        this.options.setCodecName("camel");
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        consumer = vertx.eventBus().localConsumer(id, this);
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        if (consumer != null) {
+            consumer.unregister();
+        }
+    }
+
+    @Override
+    public void handle(Message<Exchange> event) {
+        Exchange exchange = event.body();
+        Expression expression = (Expression) exchange.removeProperty("CamelVertxExpression");
+        // TODO: execute blocking
+        transform(exchange, expression);
+        // signal we are done
+        event.reply(exchange, options);
+    }
+
+    private void transform(Exchange exchange, Expression expression) {
+        Object newBody = expression.evaluate(exchange, Object.class);
+
+        if (exchange.getException() != null) {
+            // the expression threw an exception so we should break-out
+            return;
+        }
+
+        boolean out = exchange.hasOut();
+        org.apache.camel.Message old = out ? exchange.getOut() : exchange.getIn();
+
+        // create a new message container so we do not drag specialized message objects along
+        // but that is only needed if the old message is a specialized message
+        boolean copyNeeded = !(old.getClass().equals(DefaultMessage.class));
+
+        if (copyNeeded) {
+            org.apache.camel.Message msg = new DefaultMessage();
+            msg.copyFrom(old);
+            msg.setBody(newBody);
+
+            // replace message on exchange (must set as OUT)
+            ExchangeHelper.replaceMessage(exchange, msg, true);
+        } else {
+            // no copy needed so set replace value directly
+            old.setBody(newBody);
+
+            // but the message must be on OUT
+            if (!exchange.hasOut()) {
+                exchange.setOut(exchange.getIn());
+            }
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/3e3e1803/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxProcessorFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxProcessorFactory.java
b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxProcessorFactory.java
index d335725..6131d34 100644
--- a/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxProcessorFactory.java
+++ b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxProcessorFactory.java
@@ -17,9 +17,11 @@
 package org.apache.camel.component.vertx.eventbus;
 
 import io.vertx.core.Vertx;
+import org.apache.camel.Expression;
 import org.apache.camel.Processor;
 import org.apache.camel.model.ProcessorDefinition;
 import org.apache.camel.model.ToDefinition;
+import org.apache.camel.model.TransformDefinition;
 import org.apache.camel.spi.ProcessorFactory;
 import org.apache.camel.spi.RouteContext;
 
@@ -43,6 +45,9 @@ public class VertxProcessorFactory implements ProcessorFactory {
         if (def instanceof ToDefinition) {
             String uri = ((ToDefinition) def).getEndpointUri();
             return new VertxSendToProcessor(vertx, id, uri);
+        } else if (def instanceof TransformDefinition) {
+            Expression expression = ((TransformDefinition) def).getExpression().createExpression(routeContext);
+            return new VertxTransformProcessor(vertx, id, expression);
         }
 
         throw new UnsupportedOperationException("EIP not supported yet");

http://git-wip-us.apache.org/repos/asf/camel/blob/3e3e1803/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxSendToProcessor.java
----------------------------------------------------------------------
diff --git a/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxSendToProcessor.java
b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxSendToProcessor.java
index 139a289..c99dd6a 100644
--- a/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxSendToProcessor.java
+++ b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxSendToProcessor.java
@@ -18,10 +18,12 @@ package org.apache.camel.component.vertx.eventbus;
 
 import io.vertx.core.Vertx;
 import io.vertx.core.eventbus.DeliveryOptions;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
 import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
+import org.apache.camel.util.AsyncProcessorHelper;
 
-public class VertxSendToProcessor implements Processor {
+public class VertxSendToProcessor implements AsyncProcessor {
 
     private final Vertx vertx;
     private final String id;
@@ -37,9 +39,21 @@ public class VertxSendToProcessor implements Processor {
     }
 
     @Override
-    public void process(Exchange exchange) throws Exception {
+    public boolean process(final Exchange exchange, final AsyncCallback callback) {
         // if OUT then use reply handler to update exchange with result
         exchange.setProperty("CamelVertxUrl", uri);
-        vertx.eventBus().send(VertxCamelProducer.class.getName(), exchange, options);
+        vertx.eventBus().send(VertxCamelProducer.class.getName(), exchange, options, (handler)
-> {
+            if (handler.failed()) {
+                Throwable t = handler.cause();
+                exchange.setException(t);
+            }
+            callback.done(false);
+        });
+        return false;
+    }
+
+    @Override
+    public void process(Exchange exchange) throws Exception {
+        AsyncProcessorHelper.process(this, exchange);
     }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/3e3e1803/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxTransformProcessor.java
----------------------------------------------------------------------
diff --git a/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxTransformProcessor.java
b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxTransformProcessor.java
new file mode 100644
index 0000000..368b12c
--- /dev/null
+++ b/components/camel-vertx/src/main/java/org/apache/camel/component/vertx/eventbus/VertxTransformProcessor.java
@@ -0,0 +1,60 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.vertx.eventbus;
+
+import io.vertx.core.Vertx;
+import io.vertx.core.eventbus.DeliveryOptions;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.util.AsyncProcessorHelper;
+
+public class VertxTransformProcessor implements AsyncProcessor {
+
+    private final Vertx vertx;
+    private final String id;
+    private final Expression expression;
+    private final DeliveryOptions options;
+
+    public VertxTransformProcessor(Vertx vertx, String id, Expression expression) {
+        this.vertx = vertx;
+        this.id = id;
+        this.expression = expression;
+        this.options = new DeliveryOptions();
+        this.options.setCodecName("camel");
+    }
+
+    @Override
+    public void process(Exchange exchange) throws Exception {
+        AsyncProcessorHelper.process(this, exchange);
+    }
+
+    @Override
+    public boolean process(final Exchange exchange, final AsyncCallback callback) {
+        exchange.setProperty("CamelVertxExpression", expression);
+        vertx.eventBus().send(VertxCamelTransform.class.getName(), exchange, options, (handler)
-> {
+            if (handler.failed()) {
+                Throwable t = handler.cause();
+                exchange.setException(t);
+            }
+            callback.done(false);
+        });
+        return false;
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/3e3e1803/components/camel-vertx/src/test/java/org/apache/camel/component/vertx/eventbus/VertxTransformTest.java
----------------------------------------------------------------------
diff --git a/components/camel-vertx/src/test/java/org/apache/camel/component/vertx/eventbus/VertxTransformTest.java
b/components/camel-vertx/src/test/java/org/apache/camel/component/vertx/eventbus/VertxTransformTest.java
new file mode 100644
index 0000000..c29fd48
--- /dev/null
+++ b/components/camel-vertx/src/test/java/org/apache/camel/component/vertx/eventbus/VertxTransformTest.java
@@ -0,0 +1,71 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.vertx.eventbus;
+
+import io.vertx.core.Vertx;
+import org.apache.camel.CamelContext;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.vertx.VertxBaseTestSupport;
+import org.junit.Test;
+
+public class VertxTransformTest extends VertxBaseTestSupport {
+
+    private Vertx vertx;
+
+    @Override
+    protected CamelContext createCamelContext() throws Exception {
+        CamelContext context = super.createCamelContext();
+
+        vertx = Vertx.vertx();
+        vertx.eventBus().registerCodec(new VertxExchangeCodec());
+
+        VertxProcessorFactory pf = new VertxProcessorFactory(vertx);
+        context.setProcessorFactory(pf);
+
+        VertxCamelProducer vcp = new VertxCamelProducer(context, vertx, VertxCamelProducer.class.getName());
+        context.addService(vcp);
+
+        VertxCamelTransform vct = new VertxCamelTransform(context, vertx, VertxCamelTransform.class.getName());
+        context.addService(vct);
+
+        return context;
+    }
+
+    @Test
+    public void testVertxTransform() throws Exception {
+        getMockEndpoint("mock:foo").expectedBodiesReceived("World");
+        getMockEndpoint("mock:bar").expectedBodiesReceived("Hello World");
+
+        template.sendBody("direct:start", "World");
+
+        assertMockEndpointsSatisfied();
+    }
+
+    @Override
+    protected RoutesBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                from("direct:start")
+                    .to("mock:foo")
+                    .transform(simple("Hello ${body}"))
+                    .to("mock:bar");
+            }
+        };
+    }
+}


Mime
View raw message