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 <jvazq...@tecsisa.com> Authored: Wed Aug 27 19:36:31 2014 +0200 Committer: Claus Ibsen <davscl...@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)); + } + }; + } + +}
