camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [2/4] camel git commit: publish topic operation support added.
Date Wed, 03 Dec 2014 19:01:50 GMT
publish topic operation support added.


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

Branch: refs/heads/master
Commit: 9e5968c3adeacde3fad5b5a9a11572ef83b3538e
Parents: 9ecc0cc
Author: juanjovazquez <jvazquez@tecsisa.com>
Authored: Wed Aug 27 19:36:31 2014 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Wed Dec 3 19:57:21 2014 +0100

----------------------------------------------------------------------
 .../component/hazelcast/HazelcastComponent.java |  7 ++
 .../hazelcast/HazelcastComponentHelper.java     |  3 +
 .../component/hazelcast/HazelcastConstants.java |  4 +
 .../hazelcast/topic/HazelcastTopicEndpoint.java | 46 +++++++++++
 .../hazelcast/topic/HazelcastTopicProducer.java | 64 ++++++++++++++++
 .../hazelcast/HazelcastTopicProducerTest.java   | 80 ++++++++++++++++++++
 6 files changed, 204 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/9e5968c3/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastComponent.java
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastComponent.java
index a95ccda..fe21892 100644
--- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastComponent.java
+++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastComponent.java
@@ -32,6 +32,7 @@ import org.apache.camel.component.hazelcast.multimap.HazelcastMultimapEndpoint;
 import org.apache.camel.component.hazelcast.queue.HazelcastQueueEndpoint;
 import org.apache.camel.component.hazelcast.seda.HazelcastSedaConfiguration;
 import org.apache.camel.component.hazelcast.seda.HazelcastSedaEndpoint;
+import org.apache.camel.component.hazelcast.topic.HazelcastTopicEndpoint;
 import org.apache.camel.impl.DefaultComponent;
 
 import static org.apache.camel.util.ObjectHelper.removeStartingCharacters;
@@ -93,6 +94,12 @@ public class HazelcastComponent extends DefaultComponent {
             endpoint = new HazelcastQueueEndpoint(hzInstance, uri, this, remaining);
         }
 
+        if (remaining.startsWith(HazelcastConstants.TOPIC_PREFIX)) {
+            // remaining is anything (name it foo ;)
+            remaining = removeStartingCharacters(remaining.substring(HazelcastConstants.TOPIC_PREFIX.length()),
'/');
+            endpoint = new HazelcastTopicEndpoint(hzInstance, uri, this, remaining);
+        }
+
         if (remaining.startsWith(HazelcastConstants.SEDA_PREFIX)) {
             final HazelcastSedaConfiguration config = new HazelcastSedaConfiguration();
             setProperties(config, parameters);

http://git-wip-us.apache.org/repos/asf/camel/blob/9e5968c3/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastComponentHelper.java
----------------------------------------------------------------------
diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastComponentHelper.java
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastComponentHelper.java
index b1e25ee..f6057c5 100644
--- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastComponentHelper.java
+++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastComponentHelper.java
@@ -107,6 +107,9 @@ public final class HazelcastComponentHelper {
         addMapping("offer", HazelcastConstants.OFFER_OPERATION);
         addMapping("peek", HazelcastConstants.PEEK_OPERATION);
         addMapping("poll", HazelcastConstants.POLL_OPERATION);
+
+        // topic
+        addMapping("publish", HazelcastConstants.PUBLISH_OPERATION);
     }
 
     private void addMapping(String operationName, int operationNumber) {

http://git-wip-us.apache.org/repos/asf/camel/blob/9e5968c3/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastConstants.java
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastConstants.java
index fa94b01..ebd20f6 100644
--- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastConstants.java
+++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastConstants.java
@@ -26,6 +26,7 @@ public final class HazelcastConstants {
     public static final String ATOMICNUMBER_PREFIX = "atomicvalue:";
     public static final String INSTANCE_PREFIX = "instance:";
     public static final String QUEUE_PREFIX = "queue:";
+    public static final String TOPIC_PREFIX = "topic:";
     public static final String SEDA_PREFIX = "seda:";
     public static final String LIST_PREFIX = "list:";
 
@@ -70,6 +71,9 @@ public final class HazelcastConstants {
     public static final int PEEK_OPERATION = 33;
     public static final int POLL_OPERATION = 34;
 
+    // topic
+    public static final int PUBLISH_OPERATION = 35;
+
     /*
      * header values
      */

http://git-wip-us.apache.org/repos/asf/camel/blob/9e5968c3/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicEndpoint.java
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicEndpoint.java
new file mode 100644
index 0000000..34a801e
--- /dev/null
+++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicEndpoint.java
@@ -0,0 +1,46 @@
+/**
+ * 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.hazelcast.topic;
+
+import com.hazelcast.core.HazelcastInstance;
+import org.apache.camel.Component;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.hazelcast.HazelcastDefaultEndpoint;
+
+/**
+ *
+ */
+public class HazelcastTopicEndpoint extends HazelcastDefaultEndpoint {
+
+    public HazelcastTopicEndpoint(HazelcastInstance hazelcastInstance, String endpointUri,
Component component, String cacheName) {
+        super(hazelcastInstance, endpointUri, component, cacheName);
+    }
+
+    @Override
+    public Consumer createConsumer(Processor processor) throws Exception {
+        // TODO
+        return null;
+    }
+
+    @Override
+    public Producer createProducer() throws Exception {
+        return new HazelcastTopicProducer(hazelcastInstance, this, cacheName);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/9e5968c3/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicProducer.java
b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicProducer.java
new file mode 100644
index 0000000..6111c59
--- /dev/null
+++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicProducer.java
@@ -0,0 +1,64 @@
+/**
+ * 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.hazelcast.topic;
+
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.ITopic;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.hazelcast.HazelcastComponentHelper;
+import org.apache.camel.component.hazelcast.HazelcastConstants;
+import org.apache.camel.component.hazelcast.HazelcastDefaultEndpoint;
+import org.apache.camel.component.hazelcast.HazelcastDefaultProducer;
+
+/**
+ *
+ */
+public class HazelcastTopicProducer extends HazelcastDefaultProducer {
+
+    private ITopic<Object> topic;
+
+    public HazelcastTopicProducer(HazelcastInstance hazelcastInstance, HazelcastDefaultEndpoint
endpoint, String topicName) {
+        super(endpoint);
+        this.topic = hazelcastInstance.getTopic(topicName);
+    }
+
+    public void process(Exchange exchange) throws Exception {
+
+        final int operation = lookupOperationNumber(exchange);
+
+        switch (operation) {
+
+            case -1:
+                // default operation to publish
+            case HazelcastConstants.PUBLISH_OPERATION:
+                this.publish(exchange);
+                break;
+            default:
+                throw new IllegalArgumentException(String.format("The value '%s' is not allowed
for parameter '%s' on the TOPIC cache.", operation, HazelcastConstants.OPERATION));
+
+        }
+
+        // finally copy headers
+        HazelcastComponentHelper.copyHeaders(exchange);
+
+    }
+
+    private void publish(Exchange exchange) {
+        Object body = exchange.getIn().getBody();
+        this.topic.publish(body);
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/9e5968c3/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastTopicProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastTopicProducerTest.java
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastTopicProducerTest.java
new file mode 100644
index 0000000..240f5c6
--- /dev/null
+++ b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastTopicProducerTest.java
@@ -0,0 +1,80 @@
+/**
+ * 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.hazelcast;
+
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.ITopic;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.After;
+import org.junit.Test;
+import org.mockito.Mock;
+
+import static org.mockito.Mockito.*;
+
+public class HazelcastTopicProducerTest extends HazelcastCamelTestSupport {
+
+    @Mock
+    private ITopic<String> topic;
+
+    @Override
+    protected void trainHazelcastInstance(HazelcastInstance hazelcastInstance) {
+        when(hazelcastInstance.<String>getTopic("bar")).thenReturn(topic);
+    }
+
+    @Override
+    protected void verifyHazelcastInstance(HazelcastInstance hazelcastInstance) {
+        verify(hazelcastInstance, atLeastOnce()).getTopic("bar");
+    }
+
+    @After
+    public void verifyQueueMock() {
+        verifyNoMoreInteractions(topic);
+    }
+
+    @Test(expected = CamelExecutionException.class)
+    public void testWithInvalidOperation() {
+        template.sendBody("direct:publishInvalid", "foo");
+    }
+
+    @Test
+    public void noOperation() {
+        template.sendBody("direct:no-operation", "bar");
+        verify(topic).publish("bar");
+    }
+
+    @Test
+    public void add() {
+        template.sendBody("direct:publish", "bar");
+        verify(topic).publish("bar");
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                from("direct:no-operation").to(String.format("hazelcast:%sbar", HazelcastConstants.TOPIC_PREFIX));
+
+                from("direct:publishInvalid").setHeader(HazelcastConstants.OPERATION, constant("bogus")).to(String.format("hazelcast:%sbar",
HazelcastConstants.TOPIC_PREFIX));
+
+                from("direct:publish").setHeader(HazelcastConstants.OPERATION, constant(HazelcastConstants.PUBLISH_OPERATION)).to(String.format("hazelcast:%sbar",
HazelcastConstants.TOPIC_PREFIX));
+            }
+        };
+    }
+
+}


Mime
View raw message