You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ra...@apache.org on 2015/12/23 16:57:02 UTC
[1/6] camel git commit: CAMEL-9445 camel-ignite: New Apache Ignite
component. Initial work.
Repository: camel
Updated Branches:
refs/heads/master d20d0be4c -> 0350423d9
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteComputeTest.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteComputeTest.java b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteComputeTest.java
new file mode 100644
index 0000000..fd66450
--- /dev/null
+++ b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteComputeTest.java
@@ -0,0 +1,192 @@
+/**
+ * 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.ignite;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.events.EventType;
+import org.junit.After;
+import org.junit.Test;
+
+import static com.google.common.truth.Truth.assert_;
+
+public class IgniteComputeTest extends AbstractIgniteTest {
+
+ private static final List<Ignite> ADDITIONAL_INSTANCES = Lists.newArrayList();
+ private static final List<UUID> LISTENERS = Lists.newArrayList();
+
+ @Test
+ public void testExecuteWithWrongPayload() {
+ try {
+ template.requestBody("ignite:compute:abc?executionType=EXECUTE", TestIgniteComputeResources.TEST_CALLABLE, String.class);
+ } catch (Exception e) {
+ assert_().that(ObjectHelper.getException(RuntimeCamelException.class, e).getMessage()).startsWith("Ignite Compute endpoint with EXECUTE");
+ return;
+ }
+
+ fail();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testCall() {
+ TestIgniteComputeResources.COUNTER.set(0);
+
+ // Single Callable.
+ String result = template.requestBody("ignite:compute:abc?executionType=CALL", TestIgniteComputeResources.TEST_CALLABLE, String.class);
+
+ assert_().that(result).isEqualTo("hello");
+
+ // Collection of Callables.
+ Object[] callables = new Object[5];
+ Arrays.fill(callables, TestIgniteComputeResources.TEST_CALLABLE);
+ Collection<String> colResult = template.requestBody("ignite:compute:abc?executionType=CALL", Lists.newArrayList(callables), Collection.class);
+
+ assert_().that(colResult).containsExactly("hello", "hello", "hello", "hello", "hello").inOrder();
+
+ // Callables with a Reducer.
+ String reduced = template.requestBodyAndHeader("ignite:compute:abc?executionType=CALL", Lists.newArrayList(callables), IgniteConstants.IGNITE_COMPUTE_REDUCER,
+ TestIgniteComputeResources.STRING_JOIN_REDUCER, String.class);
+
+ assert_().that(reduced).isEqualTo("hellohellohellohellohello");
+ }
+
+ @Test
+ public void testRun() {
+ TestIgniteComputeResources.COUNTER.set(0);
+
+ // Single Runnable.
+ Object result = template.requestBody("ignite:compute:abc?executionType=RUN", TestIgniteComputeResources.TEST_RUNNABLE_COUNTER, Object.class);
+ assert_().that(result).isNull();
+ assert_().that(TestIgniteComputeResources.COUNTER.get()).isEqualTo(1);
+
+ // Multiple Runnables.
+ Object[] runnables = new Object[5];
+ Arrays.fill(runnables, TestIgniteComputeResources.TEST_RUNNABLE_COUNTER);
+ result = template.requestBody("ignite:compute:abc?executionType=RUN", Lists.newArrayList(runnables), Collection.class);
+ assert_().that(result).isNull();
+ assert_().that(TestIgniteComputeResources.COUNTER.get()).isEqualTo(6);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testBroadcast() {
+ TestIgniteComputeResources.COUNTER.set(0);
+
+ startAdditionalGridInstance();
+ startAdditionalGridInstance();
+
+ ignite().events().enableLocal(EventType.EVT_JOB_FINISHED);
+ LISTENERS.add(ignite().events().remoteListen(null, TestIgniteComputeResources.EVENT_COUNTER, EventType.EVT_JOB_FINISHED));
+
+ // Single Runnable.
+ Object result = template.requestBody("ignite:compute:abc?executionType=BROADCAST", TestIgniteComputeResources.TEST_RUNNABLE, Object.class);
+ assert_().that(result).isNull();
+ assert_().that(TestIgniteComputeResources.COUNTER.get()).isEqualTo(3);
+
+ // Single Callable.
+ Collection<String> colResult = template.requestBody("ignite:compute:abc?executionType=BROADCAST", TestIgniteComputeResources.TEST_CALLABLE, Collection.class);
+ assert_().that(colResult).isNotNull();
+ assert_().that(colResult).containsExactly("hello", "hello", "hello").inOrder();
+
+ // Single Closure.
+ colResult = template.requestBodyAndHeader("ignite:compute:abc?executionType=BROADCAST", TestIgniteComputeResources.TEST_CLOSURE, IgniteConstants.IGNITE_COMPUTE_PARAMS, "Camel",
+ Collection.class);
+ assert_().that(colResult).isNotNull();
+ assert_().that(colResult).containsExactly("hello Camel", "hello Camel", "hello Camel").inOrder();
+ }
+
+ @Test
+ public void testExecute() {
+ TestIgniteComputeResources.COUNTER.set(0);
+
+ startAdditionalGridInstance();
+ startAdditionalGridInstance();
+
+ ignite().events().enableLocal(EventType.EVT_JOB_RESULTED);
+ LISTENERS.add(ignite().events().remoteListen(null, TestIgniteComputeResources.EVENT_COUNTER, EventType.EVT_JOB_RESULTED));
+
+ // ComputeTask instance.
+ String result = template.requestBodyAndHeader("ignite:compute:abc?executionType=EXECUTE", TestIgniteComputeResources.COMPUTE_TASK, IgniteConstants.IGNITE_COMPUTE_PARAMS, 10, String.class);
+ assert_().that(result).isNotNull();
+ assert_().that(Splitter.on(",").splitToList(result)).containsAllOf("a0", "a1", "a2", "a3", "a4", "a5", "a6", "a7", "a8", "a9");
+
+ // ComputeTask class.
+ result = template.requestBodyAndHeader("ignite:compute:abc?executionType=EXECUTE", TestIgniteComputeResources.COMPUTE_TASK.getClass(), IgniteConstants.IGNITE_COMPUTE_PARAMS, 10, String.class);
+ assert_().that(result).isNotNull();
+ assert_().that(Splitter.on(",").splitToList(result)).containsAllOf("a0", "a1", "a2", "a3", "a4", "a5", "a6", "a7", "a8", "a9");
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testApply() {
+ TestIgniteComputeResources.COUNTER.set(0);
+
+ // Closure with a single parameter.
+ String result = template.requestBodyAndHeader("ignite:compute:abc?executionType=APPLY", TestIgniteComputeResources.TEST_CLOSURE, IgniteConstants.IGNITE_COMPUTE_PARAMS, "Camel", String.class);
+ assert_().that(result).isEqualTo("hello Camel");
+
+ // Closure with a Collection of parameters.
+ Collection<String> colResult = template.requestBodyAndHeader("ignite:compute:abc?executionType=APPLY", TestIgniteComputeResources.TEST_CLOSURE, IgniteConstants.IGNITE_COMPUTE_PARAMS,
+ Lists.newArrayList("Camel1", "Camel2", "Camel3"), Collection.class);
+ assert_().that(colResult).containsAllOf("hello Camel1", "hello Camel2", "hello Camel3");
+
+ // Closure with a Collection of parameters and a Reducer.
+ Map<String, Object> headers = ImmutableMap.<String, Object> of(IgniteConstants.IGNITE_COMPUTE_PARAMS, Lists.newArrayList("Camel1", "Camel2", "Camel3"), IgniteConstants.IGNITE_COMPUTE_REDUCER,
+ TestIgniteComputeResources.STRING_JOIN_REDUCER);
+ result = template.requestBodyAndHeaders("ignite:compute:abc?executionType=APPLY", TestIgniteComputeResources.TEST_CLOSURE, headers, String.class);
+ assert_().that(result).isEqualTo("hello Camel1hello Camel2hello Camel3");
+ }
+
+ @Override
+ public boolean isCreateCamelContextPerClass() {
+ return true;
+ }
+
+ private void startAdditionalGridInstance() {
+ ADDITIONAL_INSTANCES.add(Ignition.start(buildComponent().getIgniteConfiguration()));
+ }
+
+ @After
+ public void stopAdditionalIgniteInstances() {
+ for (Ignite ignite : ADDITIONAL_INSTANCES) {
+ ignite.close();
+ }
+ ADDITIONAL_INSTANCES.clear();
+ }
+
+ @After
+ public void stopRemoteListeners() {
+ for (UUID uuid : LISTENERS) {
+ ignite().events().stopRemoteListen(uuid);
+ }
+ LISTENERS.clear();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteEventsTest.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteEventsTest.java b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteEventsTest.java
new file mode 100644
index 0000000..316eacd
--- /dev/null
+++ b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteEventsTest.java
@@ -0,0 +1,168 @@
+/**
+ * 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.ignite;
+
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import javax.cache.expiry.CreatedExpiryPolicy;
+import javax.cache.expiry.Duration;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Route;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.events.EventType;
+import org.junit.After;
+import org.junit.Test;
+
+import static com.google.common.truth.Truth.assert_;
+
+public class IgniteEventsTest extends AbstractIgniteTest {
+
+ @Test
+ public void testConsumeAllEvents() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("ignite:events:abc").to("mock:test1");
+ }
+ });
+
+ getMockEndpoint("mock:test1").expectedMinimumMessageCount(9);
+
+ IgniteCache<String, String> cache = ignite().getOrCreateCache("abc");
+
+ // Generate cache activity.
+ cache.put("abc", "123");
+ cache.get("abc");
+ cache.remove("abc");
+ cache.withExpiryPolicy(CreatedExpiryPolicy.factoryOf(new Duration(TimeUnit.MILLISECONDS, 100)).create()).put("abc", "123");
+
+ Thread.sleep(150);
+
+ cache.get("abc");
+
+ assertMockEndpointsSatisfied();
+
+ List<Integer> eventTypes = receivedEventTypes("mock:test1");
+
+ assert_().that(eventTypes).containsAllOf(EventType.EVT_CACHE_STARTED, EventType.EVT_CACHE_ENTRY_CREATED, EventType.EVT_CACHE_OBJECT_PUT, EventType.EVT_CACHE_OBJECT_READ,
+ EventType.EVT_CACHE_OBJECT_REMOVED, EventType.EVT_CACHE_OBJECT_PUT, EventType.EVT_CACHE_OBJECT_EXPIRED, EventType.EVT_CACHE_OBJECT_READ).inOrder();
+
+ }
+
+ @Test
+ public void testConsumeFilteredEventsWithRef() throws Exception {
+ context.getRegistry(JndiRegistry.class).bind("filter", Sets.newHashSet(EventType.EVT_CACHE_OBJECT_PUT));
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("ignite:events:abc?events=#filter").to("mock:test2");
+ }
+ });
+
+ getMockEndpoint("mock:test2").expectedMessageCount(2);
+
+ IgniteCache<String, String> cache = ignite().getOrCreateCache("abc");
+
+ // Generate cache activity.
+ cache.put("abc", "123");
+ cache.get("abc");
+ cache.remove("abc");
+ cache.get("abc");
+ cache.put("abc", "123");
+
+ assertMockEndpointsSatisfied();
+
+ List<Integer> eventTypes = receivedEventTypes("mock:test2");
+
+ assert_().that(eventTypes).containsExactly(EventType.EVT_CACHE_OBJECT_PUT, EventType.EVT_CACHE_OBJECT_PUT).inOrder();
+ }
+
+ @Test
+ public void testConsumeFilteredEventsInline() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("ignite:events:abc?events=EVT_CACHE_OBJECT_PUT").to("mock:test3");
+ }
+ });
+
+ getMockEndpoint("mock:test3").expectedMessageCount(2);
+
+ IgniteCache<String, String> cache = ignite().getOrCreateCache("abc");
+
+ // Generate cache activity.
+ cache.put("abc", "123");
+ cache.get("abc");
+ cache.remove("abc");
+ cache.get("abc");
+ cache.put("abc", "123");
+
+ assertMockEndpointsSatisfied();
+
+ List<Integer> eventTypes = receivedEventTypes("mock:test3");
+
+ assert_().that(eventTypes).containsExactly(EventType.EVT_CACHE_OBJECT_PUT, EventType.EVT_CACHE_OBJECT_PUT).inOrder();
+
+ }
+
+ private List<Integer> receivedEventTypes(String mockEndpoint) {
+ List<Integer> eventTypes = Lists.newArrayList(Lists.transform(getMockEndpoint(mockEndpoint).getExchanges(), new Function<Exchange, Integer>() {
+ @Override
+ public Integer apply(Exchange input) {
+ return input.getIn().getBody(Event.class).type();
+ }
+ }));
+ return eventTypes;
+ }
+
+ @Override
+ public boolean isCreateCamelContextPerClass() {
+ return true;
+ }
+
+ @After
+ public void stopAllRoutes() throws Exception {
+ for (Route route : context.getRoutes()) {
+ if (context.getRouteStatus(route.getId()) != ServiceStatus.Started) {
+ return;
+ }
+ context.stopRoute(route.getId());
+ }
+ resetMocks();
+ }
+
+ @Override
+ protected IgniteComponent buildComponent() {
+ IgniteConfiguration config = new IgniteConfiguration();
+ config.setIncludeEventTypes(EventType.EVTS_ALL_MINUS_METRIC_UPDATE);
+ return IgniteComponent.fromConfiguration(config);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteIdGenTest.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteIdGenTest.java b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteIdGenTest.java
new file mode 100644
index 0000000..c310e41
--- /dev/null
+++ b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteIdGenTest.java
@@ -0,0 +1,83 @@
+/**
+ * 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.ignite;
+
+import com.google.common.collect.ImmutableSet;
+
+import org.apache.camel.component.ignite.idgen.IgniteIdGenEndpoint;
+import org.apache.camel.component.ignite.idgen.IgniteIdGenOperation;
+import org.apache.ignite.IgniteAtomicSequence;
+import org.junit.After;
+import org.junit.Test;
+
+import static com.google.common.truth.Truth.assert_;
+
+public class IgniteIdGenTest extends AbstractIgniteTest {
+
+ @Test
+ public void testOperations() {
+ assert_().that(template.requestBody("ignite:idgen:abc?initialValue=0&operation=GET", null, Long.class)).isEqualTo(0);
+ assert_().that(template.requestBody("ignite:idgen:abc?initialValue=0&operation=GET_AND_INCREMENT", null, Long.class)).isEqualTo(0);
+ assert_().that(template.requestBody("ignite:idgen:abc?initialValue=0&operation=INCREMENT_AND_GET", null, Long.class)).isEqualTo(2);
+ assert_().that(template.requestBody("ignite:idgen:abc?initialValue=0&operation=ADD_AND_GET", 5, Long.class)).isEqualTo(7);
+ assert_().that(template.requestBody("ignite:idgen:abc?initialValue=0&operation=GET_AND_ADD", 5, Long.class)).isEqualTo(7);
+ assert_().that(template.requestBody("ignite:idgen:abc?initialValue=0&operation=GET", 5, Long.class)).isEqualTo(12);
+ }
+
+ @Test
+ public void testInitialValue() {
+ assert_().that(template.requestBody("ignite:idgen:abc?operation=GET&initialValue=100", null, Long.class)).isEqualTo(100);
+ assert_().that(template.requestBody("ignite:idgen:abc?operation=GET_AND_INCREMENT&initialValue=100", null, Long.class)).isEqualTo(100);
+ assert_().that(template.requestBody("ignite:idgen:abc?operation=INCREMENT_AND_GET&initialValue=100", null, Long.class)).isEqualTo(102);
+ assert_().that(template.requestBody("ignite:idgen:abc?operation=ADD_AND_GET&initialValue=100", 5, Long.class)).isEqualTo(107);
+ assert_().that(template.requestBody("ignite:idgen:abc?operation=GET_AND_ADD&initialValue=100", 5, Long.class)).isEqualTo(107);
+ assert_().that(template.requestBody("ignite:idgen:abc?operation=GET&initialValue=100", 5, Long.class)).isEqualTo(112);
+ }
+
+ @Test
+ public void testDifferentOperation() {
+ assert_().that(template.requestBody("ignite:idgen:abc?operation=GET&initialValue=100", null, Long.class)).isEqualTo(100);
+ assert_().that(template.requestBodyAndHeader("ignite:idgen:abc?operation=GET_AND_INCREMENT&initialValue=100", null, IgniteConstants.IGNITE_IDGEN_OPERATION,
+ IgniteIdGenOperation.INCREMENT_AND_GET, Long.class)).isEqualTo(101);
+ }
+
+ @Test
+ public void testBatchSize() {
+ IgniteIdGenEndpoint endpoint = context.getEndpoint("ignite:idgen:abc?operation=GET&initialValue=100&batchSize=100", IgniteIdGenEndpoint.class);
+ assert_().that(template.requestBody(endpoint, null, Long.class)).isEqualTo(100);
+
+ // Cannot test much here with a single Ignite instance, let's just test that the parameter could be set.
+ assert_().that(endpoint.getBatchSize());
+ }
+
+ @Override
+ public boolean isCreateCamelContextPerClass() {
+ return true;
+ }
+
+ @After
+ public void deleteSets() {
+ for (String name : ImmutableSet.<String> of("abc")) {
+ IgniteAtomicSequence seq = ignite().atomicSequence(name, 0, false);
+ if (seq == null) {
+ continue;
+ }
+ seq.close();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteMessagingTest.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteMessagingTest.java b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteMessagingTest.java
new file mode 100644
index 0000000..a1cee04
--- /dev/null
+++ b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteMessagingTest.java
@@ -0,0 +1,164 @@
+/**
+ * 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.ignite;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ContiguousSet;
+import com.google.common.collect.DiscreteDomain;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Range;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.junit.After;
+import org.junit.Test;
+
+import static com.google.common.truth.Truth.assert_;
+import static com.jayway.awaitility.Awaitility.await;
+import static com.jayway.awaitility.Awaitility.to;
+import static org.hamcrest.Matchers.equalTo;
+
+public class IgniteMessagingTest extends AbstractIgniteTest implements Serializable {
+
+ private static final long serialVersionUID = 3967738538216977749L;
+
+ private static final String TOPIC1 = "TOPIC1";
+ private static final String TOPIC2 = "TOPIC2";
+ private UUID uuid;
+
+ @Test
+ public void testProducerSendMessage() {
+ List<Object> messages = Lists.newArrayList();
+ setupMessageListener(TOPIC1, messages);
+
+ template.requestBody("ignite:messaging:TOPIC1", 1);
+
+ await().atMost(5, TimeUnit.SECONDS).untilCall(to(messages).size(), equalTo(1));
+ assert_().that(messages.get(0)).isEqualTo(1);
+ }
+
+ @Test
+ public void testProducerSendMessageTopicInHeader() throws Exception {
+ List<Object> messages1 = Lists.newArrayList();
+ setupMessageListener(TOPIC1, messages1);
+
+ List<Object> messages2 = Lists.newArrayList();
+ setupMessageListener(TOPIC2, messages2);
+
+ template.requestBodyAndHeader("ignite:messaging:TOPIC1", 1, IgniteConstants.IGNITE_MESSAGING_TOPIC, "TOPIC2");
+
+ Thread.sleep(1000);
+ assert_().that(messages1.size()).isEqualTo(0);
+ assert_().that(messages2.size()).isEqualTo(1);
+ }
+
+ @Test
+ public void testProducerSendManyMessages() {
+ List<Object> messages = Lists.newArrayList();
+ setupMessageListener(TOPIC1, messages);
+
+ Set<Integer> request = ContiguousSet.create(Range.closedOpen(0, 100), DiscreteDomain.integers());
+ template.requestBody("ignite:messaging:TOPIC1", request);
+
+ await().atMost(5, TimeUnit.SECONDS).untilCall(to(messages).size(), equalTo(100));
+ assert_().that(messages).containsAllIn(request);
+ }
+
+ @Test
+ public void testProducerSendManyMessagesOrdered() {
+ List<Object> messages = Lists.newArrayList();
+ setupMessageListener(TOPIC1, messages);
+
+ ContiguousSet<Integer> set = ContiguousSet.create(Range.closedOpen(0, 100), DiscreteDomain.integers());
+ for (int i : set) {
+ template.requestBody("ignite:messaging:TOPIC1?sendMode=ORDERED&timeout=1000", i);
+ }
+
+ await().atMost(5, TimeUnit.SECONDS).untilCall(to(messages).size(), equalTo(100));
+ assert_().that(messages).containsAllIn(set);
+ }
+
+ @Test
+ public void testProducerSendCollectionAsObject() {
+ List<Object> messages = Lists.newArrayList();
+ setupMessageListener(TOPIC1, messages);
+
+ Set<Integer> request = ContiguousSet.create(Range.closedOpen(0, 100), DiscreteDomain.integers());
+ template.requestBody("ignite:messaging:TOPIC1?treatCollectionsAsCacheObjects=true", request);
+
+ await().atMost(5, TimeUnit.SECONDS).untilCall(to(messages).size(), equalTo(1));
+ assert_().that(messages.get(0)).isEqualTo(request);
+ }
+
+ @Test
+ public void testConsumerManyMessages() throws Exception {
+ List<Object> messages = Lists.newArrayList();
+ Consumer consumer = context.getEndpoint("ignite:messaging:TOPIC1").createConsumer(storeBodyInListProcessor(messages));
+ consumer.start();
+
+ Set<Integer> messagesToSend = ContiguousSet.create(Range.closedOpen(0, 100), DiscreteDomain.integers());
+ ignite().message().send(TOPIC1, messagesToSend);
+
+ await().atMost(5, TimeUnit.SECONDS).untilCall(to(messages).size(), equalTo(100));
+
+ consumer.stop();
+ }
+
+ private void setupMessageListener(String topic, final List<Object> messages) {
+ uuid = ignite().message().remoteListen(topic, new IgniteBiPredicate<UUID, Object>() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean apply(UUID uuid, Object message) {
+ messages.add(message);
+ return true;
+ }
+ });
+ }
+
+ @After
+ public void stopMessageListener() {
+ if (uuid == null) {
+ return;
+ }
+
+ ignite().message().stopRemoteListen(uuid);
+ uuid = null;
+ }
+
+ @Override
+ public boolean isCreateCamelContextPerClass() {
+ return true;
+ }
+
+ private Processor storeBodyInListProcessor(final List<Object> list) {
+ return new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ list.add(exchange.getIn().getBody());
+ }
+ };
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteQueueTest.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteQueueTest.java b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteQueueTest.java
new file mode 100644
index 0000000..77fb769
--- /dev/null
+++ b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteQueueTest.java
@@ -0,0 +1,257 @@
+/**
+ * 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.ignite;
+
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import org.apache.camel.component.ignite.queue.IgniteQueueEndpoint;
+import org.apache.camel.component.ignite.queue.IgniteQueueOperation;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CollectionConfiguration;
+import org.junit.After;
+import org.junit.Test;
+
+import static com.google.common.truth.Truth.assert_;
+
+
+public class IgniteQueueTest extends AbstractIgniteTest {
+
+ @Test
+ public void testOperations() {
+ boolean result = template.requestBody("ignite:queue:abc?operation=ADD", "hello", boolean.class);
+ assert_().that(result).isTrue();
+ assert_().that(ignite().queue("abc", 0, new CollectionConfiguration()).contains("hello")).isTrue();
+
+ result = template.requestBody("ignite:queue:abc?operation=CONTAINS", "hello", boolean.class);
+ assert_().that(result).isTrue();
+ assert_().that(ignite().queue("abc", 0, new CollectionConfiguration()).contains("hello")).isTrue();
+
+ result = template.requestBody("ignite:queue:abc?operation=REMOVE", "hello", boolean.class);
+ assert_().that(result).isTrue();
+ assert_().that(ignite().queue("abc", 0, new CollectionConfiguration()).contains("hello")).isFalse();
+
+ result = template.requestBody("ignite:queue:abc?operation=CONTAINS", "hello", boolean.class);
+ assert_().that(result).isFalse();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testOperations2() {
+ for (int i = 0; i < 100; i++) {
+ template.requestBody("ignite:queue:abc?operation=ADD", "hello" + i);
+ }
+
+ // SIZE
+ int size = template.requestBody("ignite:queue:abc?operation=SIZE", "hello", int.class);
+ assert_().that(size).isEqualTo(100);
+ assert_().that(ignite().queue("abc", 0, new CollectionConfiguration()).size()).isEqualTo(100);
+
+ List<String> toRetain = Lists.newArrayList();
+ for (int i = 0; i < 50; i++) {
+ toRetain.add("hello" + i);
+ }
+
+ // RETAIN_ALL
+ boolean retained = template.requestBodyAndHeader("ignite:queue:abc?operation=CLEAR", toRetain, IgniteConstants.IGNITE_QUEUE_OPERATION, IgniteQueueOperation.RETAIN_ALL, boolean.class);
+ assert_().that(retained).isTrue();
+
+ // SIZE
+ size = template.requestBody("ignite:queue:abc?operation=SIZE", "hello", int.class);
+ assert_().that(size).isEqualTo(50);
+ assert_().that(ignite().queue("abc", 0, new CollectionConfiguration()).size()).isEqualTo(50);
+
+ // ITERATOR
+ Iterator<String> iterator = template.requestBody("ignite:queue:abc?operation=ITERATOR", "hello", Iterator.class);
+ assert_().that(Iterators.toArray(iterator, String.class)).asList().containsExactlyElementsIn(toRetain).inOrder();
+
+ // ARRAY
+ String[] array = template.requestBody("ignite:queue:abc?operation=ARRAY", "hello", String[].class);
+ assert_().that(array).asList().containsExactlyElementsIn(toRetain).inOrder();
+
+ // CLEAR
+ Object result = template.requestBody("ignite:queue:abc?operation=CLEAR", "hello", String.class);
+ assert_().that(result).isEqualTo("hello");
+ assert_().that(ignite().queue("abc", 0, new CollectionConfiguration()).size()).isEqualTo(0);
+
+ // SIZE
+ size = template.requestBody("ignite:queue:abc?operation=SIZE", "hello", int.class);
+ assert_().that(size).isEqualTo(0);
+ assert_().that(ignite().queue("abc", 0, new CollectionConfiguration()).size()).isEqualTo(0);
+ }
+
+ @Test
+ public void testRetainSingle() {
+ // Fill data.
+ for (int i = 0; i < 100; i++) {
+ template.requestBody("ignite:queue:abc?operation=ADD", "hello" + i);
+ }
+
+ boolean retained = template.requestBody("ignite:queue:abc?operation=RETAIN_ALL", "hello10", boolean.class);
+ assert_().that(retained).isTrue();
+
+ // ARRAY
+ String[] array = template.requestBody("ignite:queue:abc?operation=ARRAY", "hello", String[].class);
+ assert_().that(array).asList().containsExactly("hello10");
+ }
+
+ @Test
+ public void testCollectionsAsCacheObject() {
+ // Fill data.
+ for (int i = 0; i < 100; i++) {
+ template.requestBody("ignite:queue:abc?operation=ADD", "hello" + i);
+ }
+
+ // Add the set.
+ Set<String> toAdd = Sets.newHashSet("hello101", "hello102", "hello103");
+ template.requestBody("ignite:queue:abc?operation=ADD&treatCollectionsAsCacheObjects=true", toAdd);
+
+ // Size must be 101, not 103.
+ int size = template.requestBody("ignite:queue:abc?operation=SIZE", "hello", int.class);
+ assert_().that(size).isEqualTo(101);
+ assert_().that(ignite().queue("abc", 0, new CollectionConfiguration()).size()).isEqualTo(101);
+ assert_().that(ignite().queue("abc", 0, new CollectionConfiguration()).contains(toAdd)).isTrue();
+
+ // Check whether the Set contains the Set.
+ boolean contains = template.requestBody("ignite:queue:abc?operation=CONTAINS&treatCollectionsAsCacheObjects=true", toAdd, boolean.class);
+ assert_().that(contains).isTrue();
+
+ // Delete the Set.
+ template.requestBody("ignite:queue:abc?operation=REMOVE&treatCollectionsAsCacheObjects=true", toAdd);
+
+ // Size must be 100 again.
+ size = template.requestBody("ignite:queue:abc?operation=SIZE", "hello", int.class);
+ assert_().that(size).isEqualTo(100);
+ assert_().that(ignite().queue("abc", 0, new CollectionConfiguration()).size()).isEqualTo(100);
+ assert_().that(ignite().queue("abc", 0, new CollectionConfiguration()).contains(toAdd)).isFalse();
+
+ }
+
+ @Test
+ public void testWithConfiguration() {
+ CollectionConfiguration configuration = new CollectionConfiguration();
+ configuration.setCacheMode(CacheMode.LOCAL);
+
+ context.getRegistry(JndiRegistry.class).bind("config", configuration);
+
+ IgniteQueueEndpoint igniteEndpoint = context.getEndpoint("ignite:queue:abc?operation=ADD&configuration=#config", IgniteQueueEndpoint.class);
+ template.requestBody(igniteEndpoint, "hello");
+
+ assert_().that(ignite().queue("abc", 0, configuration).size()).isEqualTo(1);
+ assert_().that(igniteEndpoint.getConfiguration()).isEqualTo(configuration);
+ }
+
+ @Test
+ public void testBoundedQueueAndOtherOperations() throws Exception {
+ List<String> list = Lists.newArrayList();
+
+ // Fill data.
+ for (int i = 0; i < 100; i++) {
+ template.requestBody("ignite:queue:def?operation=ADD&capacity=100", "hello" + i);
+ list.add("hello" + i);
+ }
+
+ // NOTE: Unfortunately the behaviour of IgniteQueue doesn't adhere to the overridden ADD method. It should return an Exception.
+ assert_().that(template.requestBody("ignite:queue:def?operation=ADD&capacity=100", "hello101", boolean.class)).isFalse();
+ assert_().that(template.requestBody("ignite:queue:def?operation=OFFER&capacity=100", "hello101", boolean.class)).isFalse();
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ Thread t = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ assert_().that(template.requestBody("ignite:queue:def?operation=PUT&capacity=100", "hello101", boolean.class)).isFalse();
+ latch.countDown();
+ }
+ });
+
+ t.start();
+
+ // Wait 2 seconds and check that the thread was blocked.
+ assert_().that(latch.await(2000, TimeUnit.MILLISECONDS)).isFalse();
+ t.interrupt();
+
+ // PEEK and ELEMENT.
+ assert_().that(template.requestBody("ignite:queue:def?operation=PEEK&capacity=100", null, String.class)).isEqualTo("hello0");
+ assert_().that(template.requestBody("ignite:queue:def?operation=ELEMENT&capacity=100", null, String.class)).isEqualTo("hello0");
+
+ // TAKE.
+ assert_().that(template.requestBody("ignite:queue:def?operation=TAKE&capacity=100", null, String.class)).isEqualTo("hello0");
+ assert_().that(template.requestBody("ignite:queue:def?operation=SIZE&capacity=100", null, int.class)).isEqualTo(99);
+
+ // Now drain.
+ assert_().that(template.requestBody("ignite:queue:def?operation=DRAIN&capacity=100", null, String[].class)).asList().hasSize(99);
+ assert_().that(template.requestBody("ignite:queue:def?operation=SIZE&capacity=100", null, int.class)).isEqualTo(0);
+ assert_().that(template.requestBody("ignite:queue:def?operation=POLL&capacity=100", null, String.class)).isNull();
+
+ // TAKE.
+ t = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ assert_().that(template.requestBody("ignite:queue:def?operation=TAKE&capacity=100", null, String.class)).isEqualTo("hello102");
+ latch.countDown();
+ }
+ });
+
+ t.start();
+
+ // Element was returned.
+ assert_().that(template.requestBody("ignite:queue:def?operation=ADD&capacity=100", "hello102", boolean.class)).isTrue();
+ assert_().that(latch.await(1000, TimeUnit.MILLISECONDS)).isTrue();
+
+ // POLL with a timeout.
+ assert_().that(Executors.newSingleThreadExecutor().submit(new Callable<Long>() {
+ @Override
+ public Long call() throws Exception {
+ Stopwatch sw = Stopwatch.createStarted();
+ assert_().that(template.requestBody("ignite:queue:def?operation=POLL&timeoutMillis=1000&capacity=100", null, String.class)).isNull();
+ return sw.elapsed(TimeUnit.MILLISECONDS);
+ }
+ }).get()).isAtLeast(1000L);
+
+ }
+
+ @Override
+ public boolean isCreateCamelContextPerClass() {
+ return true;
+ }
+
+ @After
+ public void deleteQueues() {
+ for (String queueName : ImmutableSet.<String> of("abc")) {
+ ignite().queue(queueName, 0, new CollectionConfiguration()).close();
+ }
+
+ // Bounded queues.
+ for (String queueName : ImmutableSet.<String> of("def")) {
+ ignite().queue(queueName, 100, new CollectionConfiguration()).close();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteSetTest.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteSetTest.java b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteSetTest.java
new file mode 100644
index 0000000..3c3c627
--- /dev/null
+++ b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteSetTest.java
@@ -0,0 +1,176 @@
+/**
+ * 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.ignite;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+import org.apache.camel.component.ignite.set.IgniteSetEndpoint;
+import org.apache.camel.component.ignite.set.IgniteSetOperation;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CollectionConfiguration;
+import org.junit.After;
+import org.junit.Test;
+
+import static com.google.common.truth.Truth.assert_;
+
+public class IgniteSetTest extends AbstractIgniteTest {
+
+ @Test
+ public void testOperations() {
+ boolean result = template.requestBody("ignite:set:abc?operation=ADD", "hello", boolean.class);
+ assert_().that(result).isTrue();
+ assert_().that(ignite().set("abc", new CollectionConfiguration()).contains("hello")).isTrue();
+
+ result = template.requestBody("ignite:set:abc?operation=CONTAINS", "hello", boolean.class);
+ assert_().that(result).isTrue();
+ assert_().that(ignite().set("abc", new CollectionConfiguration()).contains("hello")).isTrue();
+
+ result = template.requestBody("ignite:set:abc?operation=REMOVE", "hello", boolean.class);
+ assert_().that(result).isTrue();
+ assert_().that(ignite().set("abc", new CollectionConfiguration()).contains("hello")).isFalse();
+
+ result = template.requestBody("ignite:set:abc?operation=CONTAINS", "hello", boolean.class);
+ assert_().that(result).isFalse();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testOperations2() {
+ for (int i = 0; i < 100; i++) {
+ template.requestBody("ignite:set:abc?operation=ADD", "hello" + i);
+ }
+
+ // SIZE
+ int size = template.requestBody("ignite:set:abc?operation=SIZE", "hello", int.class);
+ assert_().that(size).isEqualTo(100);
+ assert_().that(ignite().set("abc", new CollectionConfiguration()).size()).isEqualTo(100);
+
+ List<String> toRetain = Lists.newArrayList();
+ for (int i = 0; i < 50; i++) {
+ toRetain.add("hello" + i);
+ }
+
+ // RETAIN_ALL
+ boolean retained = template.requestBodyAndHeader("ignite:set:abc?operation=CLEAR", toRetain, IgniteConstants.IGNITE_SETS_OPERATION, IgniteSetOperation.RETAIN_ALL, boolean.class);
+ assert_().that(retained).isTrue();
+
+ // SIZE
+ size = template.requestBody("ignite:set:abc?operation=SIZE", "hello", int.class);
+ assert_().that(size).isEqualTo(50);
+ assert_().that(ignite().set("abc", new CollectionConfiguration()).size()).isEqualTo(50);
+
+ // ITERATOR
+ Iterator<String> iterator = template.requestBody("ignite:set:abc?operation=ITERATOR", "hello", Iterator.class);
+ assert_().that(Iterators.toArray(iterator, String.class)).asList().containsExactlyElementsIn(toRetain);
+
+ // ARRAY
+ String[] array = template.requestBody("ignite:set:abc?operation=ARRAY", "hello", String[].class);
+ assert_().that(array).asList().containsExactlyElementsIn(toRetain);
+
+ // CLEAR
+ Object result = template.requestBody("ignite:set:abc?operation=CLEAR", "hello", String.class);
+ assert_().that(result).isEqualTo("hello");
+ assert_().that(ignite().set("abc", new CollectionConfiguration()).size()).isEqualTo(0);
+
+ // SIZE
+ size = template.requestBody("ignite:set:abc?operation=SIZE", "hello", int.class);
+ assert_().that(size).isEqualTo(0);
+ assert_().that(ignite().set("abc", new CollectionConfiguration()).size()).isEqualTo(0);
+ }
+
+ @Test
+ public void testRetainSingle() {
+ // Fill data.
+ for (int i = 0; i < 100; i++) {
+ template.requestBody("ignite:set:abc?operation=ADD", "hello" + i);
+ }
+
+ boolean retained = template.requestBody("ignite:set:abc?operation=RETAIN_ALL", "hello10", boolean.class);
+ assert_().that(retained).isTrue();
+
+ // ARRAY
+ String[] array = template.requestBody("ignite:set:abc?operation=ARRAY", "hello", String[].class);
+ assert_().that(array).asList().containsExactly("hello10");
+ }
+
+ @Test
+ public void testCollectionsAsCacheObject() {
+ // Fill data.
+ for (int i = 0; i < 100; i++) {
+ template.requestBody("ignite:set:abc?operation=ADD", "hello" + i);
+ }
+
+ // Add the set.
+ Set<String> toAdd = Sets.newHashSet("hello101", "hello102", "hello103");
+ template.requestBody("ignite:set:abc?operation=ADD&treatCollectionsAsCacheObjects=true", toAdd);
+
+ // Size must be 101, not 103.
+ int size = template.requestBody("ignite:set:abc?operation=SIZE", "hello", int.class);
+ assert_().that(size).isEqualTo(101);
+ assert_().that(ignite().set("abc", new CollectionConfiguration()).size()).isEqualTo(101);
+ assert_().that(ignite().set("abc", new CollectionConfiguration()).contains(toAdd)).isTrue();
+
+ // Check whether the Set contains the Set.
+ boolean contains = template.requestBody("ignite:set:abc?operation=CONTAINS&treatCollectionsAsCacheObjects=true", toAdd, boolean.class);
+ assert_().that(contains).isTrue();
+
+ // Delete the Set.
+ template.requestBody("ignite:set:abc?operation=REMOVE&treatCollectionsAsCacheObjects=true", toAdd);
+
+ // Size must be 100 again.
+ size = template.requestBody("ignite:set:abc?operation=SIZE", "hello", int.class);
+ assert_().that(size).isEqualTo(100);
+ assert_().that(ignite().set("abc", new CollectionConfiguration()).size()).isEqualTo(100);
+ assert_().that(ignite().set("abc", new CollectionConfiguration()).contains(toAdd)).isFalse();
+
+ }
+
+ @Test
+ public void testWithConfiguration() {
+ CollectionConfiguration configuration = new CollectionConfiguration();
+ configuration.setCacheMode(CacheMode.LOCAL);
+
+ context.getRegistry(JndiRegistry.class).bind("config", configuration);
+
+ IgniteSetEndpoint igniteEndpoint = context.getEndpoint("ignite:set:abc?operation=ADD&configuration=#config", IgniteSetEndpoint.class);
+ template.requestBody(igniteEndpoint, "hello");
+
+ assert_().that(ignite().set("abc", configuration).size()).isEqualTo(1);
+ assert_().that(igniteEndpoint.getConfiguration()).isEqualTo(configuration);
+ }
+
+ @Override
+ public boolean isCreateCamelContextPerClass() {
+ return true;
+ }
+
+ @After
+ public void deleteSets() {
+ for (String setName : ImmutableSet.<String> of("abc")) {
+ ignite().set(setName, new CollectionConfiguration()).close();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/TestIgniteComputeResources.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/TestIgniteComputeResources.java b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/TestIgniteComputeResources.java
new file mode 100644
index 0000000..7c31e16e
--- /dev/null
+++ b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/TestIgniteComputeResources.java
@@ -0,0 +1,149 @@
+/**
+ * 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.ignite;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.Lists;
+
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeTask;
+import org.apache.ignite.compute.ComputeTaskSplitAdapter;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.lang.IgniteCallable;
+import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.lang.IgniteReducer;
+import org.apache.ignite.lang.IgniteRunnable;
+
+public final class TestIgniteComputeResources {
+
+ public static final AtomicInteger COUNTER = new AtomicInteger(0);
+
+ public static final IgniteRunnable TEST_RUNNABLE = new IgniteRunnable() {
+ private static final long serialVersionUID = -4961602602993218883L;
+
+ @Override
+ public void run() {
+ System.out.println("Hello from a runnable");
+ }
+ };
+
+ public static final IgniteRunnable TEST_RUNNABLE_COUNTER = new IgniteRunnable() {
+ private static final long serialVersionUID = 386219709871673366L;
+
+ @Override
+ public void run() {
+ COUNTER.incrementAndGet();
+ }
+ };
+
+ public static final IgnitePredicate<Event> EVENT_COUNTER = new IgnitePredicate<Event>() {
+ private static final long serialVersionUID = -4214894278107593791L;
+
+ @Override
+ public boolean apply(Event event) {
+ COUNTER.incrementAndGet();
+ return true;
+ }
+ };
+
+ public static final IgniteCallable<String> TEST_CALLABLE = new IgniteCallable<String>() {
+ private static final long serialVersionUID = 986972344531961815L;
+
+ @Override
+ public String call() throws Exception {
+ return "hello";
+ }
+ };
+
+ public static final IgniteClosure<String, String> TEST_CLOSURE = new IgniteClosure<String, String>() {
+ private static final long serialVersionUID = -3969758431961263815L;
+
+ @Override
+ public String apply(String input) {
+ return "hello " + input;
+ }
+ };
+
+ public static final ComputeTask<Integer, String> COMPUTE_TASK = new ComputeTaskSplitAdapter<Integer, String>() {
+ private static final long serialVersionUID = 3040624379256407732L;
+
+ @Override
+ public String reduce(List<ComputeJobResult> results) throws IgniteException {
+ StringBuilder answer = new StringBuilder();
+ for (ComputeJobResult res : results) {
+ answer.append(res.getData()).append(",");
+ }
+ answer.deleteCharAt(answer.length() - 1);
+ return answer.toString();
+ }
+
+ @Override
+ protected Collection<? extends ComputeJob> split(int gridSize, final Integer arg) throws IgniteException {
+ Set<ComputeJob> answer = new HashSet<>();
+ for (int i = 0; i < arg; i++) {
+ final int c = i;
+ answer.add(new ComputeJob() {
+ private static final long serialVersionUID = 3365213549618276779L;
+
+ @Override
+ public Object execute() throws IgniteException {
+ return "a" + c;
+ }
+
+ @Override
+ public void cancel() {
+ // nothing
+ }
+ });
+ }
+ return answer;
+ }
+ };
+
+ public static final IgniteReducer<String, String> STRING_JOIN_REDUCER = new IgniteReducer<String, String>() {
+ private static final long serialVersionUID = 1L;
+ private List<String> list = Lists.newArrayList();
+
+ @Override
+ public boolean collect(String value) {
+ list.add(value);
+ return true;
+ }
+
+ @Override
+ public String reduce() {
+ Collections.sort(list);
+ String answer = Joiner.on("").join(list);
+ list.clear();
+ return answer;
+ }
+ };
+
+ private TestIgniteComputeResources() {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/test/resources/log4j.properties b/components/camel-ignite/src/test/resources/log4j.properties
new file mode 100644
index 0000000..cb64298
--- /dev/null
+++ b/components/camel-ignite/src/test/resources/log4j.properties
@@ -0,0 +1,37 @@
+#
+# 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.
+#
+
+log4j.rootLogger=INFO, file
+# change the logging level of this category to increase verbosity of the MongoDB component
+log4j.category.org.apache.camel.component.mongodb=INFO, file
+log4j.additivity.org.apache.camel.component.mongodb=false
+
+# uncomment the following line to turn on Camel debugging
+#log4j.logger.org.apache.camel=DEBUG
+
+# CONSOLE appender not used by default
+log4j.appender.out=org.apache.log4j.ConsoleAppender
+log4j.appender.out.layout=org.apache.log4j.PatternLayout
+log4j.appender.out.layout.ConversionPattern=[%30.30t] %-30.30c{1} %-5p %m%n
+#log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+
+
+# File appender
+log4j.appender.file=org.apache.log4j.FileAppender
+log4j.appender.file.layout=org.apache.log4j.PatternLayout
+log4j.appender.file.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+log4j.appender.file.file=target/camel-mongodb-test.log
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/pom.xml
----------------------------------------------------------------------
diff --git a/components/pom.xml b/components/pom.xml
index fe40980..a68a15e 100644
--- a/components/pom.xml
+++ b/components/pom.xml
@@ -124,6 +124,7 @@
<module>camel-hl7</module>
<module>camel-ibatis</module>
<module>camel-ical</module>
+ <module>camel-ignite</module>
<module>camel-infinispan</module>
<module>camel-irc</module>
<module>camel-jackson</module>
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/parent/pom.xml
----------------------------------------------------------------------
diff --git a/parent/pom.xml b/parent/pom.xml
index f2e3db2..2839035 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -233,6 +233,7 @@
<ibatis-bundle-version>2.3.4.726_4</ibatis-bundle-version>
<ibatis-version>2.3.4.726</ibatis-version>
<ical4j-version>1.0.7</ical4j-version>
+ <ignite-version>1.5.0-b1</ignite-version>
<infinispan-version>8.1.0.Final</infinispan-version>
<irclib-bundle-version>1.10_5</irclib-bundle-version>
<irclib-version>1.10</irclib-version>
[4/6] camel git commit: CAMEL-9445 camel-ignite: Add Karaf feature.
Posted by ra...@apache.org.
CAMEL-9445 camel-ignite: Add Karaf feature.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/d26af3dd
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/d26af3dd
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/d26af3dd
Branch: refs/heads/master
Commit: d26af3ddaa70cbe0562360b119f035467918ab96
Parents: a695c5d
Author: Raul Kripalani <ra...@apache.org>
Authored: Wed Dec 23 15:36:40 2015 +0000
Committer: Raul Kripalani <ra...@apache.org>
Committed: Wed Dec 23 15:36:40 2015 +0000
----------------------------------------------------------------------
parent/pom.xml | 1 +
platforms/karaf/features/src/main/resources/features.xml | 5 +++++
2 files changed, 6 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/d26af3dd/parent/pom.xml
----------------------------------------------------------------------
diff --git a/parent/pom.xml b/parent/pom.xml
index 2839035..8189b1a 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -234,6 +234,7 @@
<ibatis-version>2.3.4.726</ibatis-version>
<ical4j-version>1.0.7</ical4j-version>
<ignite-version>1.5.0-b1</ignite-version>
+ <ignite-version-range>[1.5.0-b1,1.6)</ignite-version-range>
<infinispan-version>8.1.0.Final</infinispan-version>
<irclib-bundle-version>1.10_5</irclib-bundle-version>
<irclib-version>1.10</irclib-version>
http://git-wip-us.apache.org/repos/asf/camel/blob/d26af3dd/platforms/karaf/features/src/main/resources/features.xml
----------------------------------------------------------------------
diff --git a/platforms/karaf/features/src/main/resources/features.xml b/platforms/karaf/features/src/main/resources/features.xml
index 2b7a26e..178180c 100644
--- a/platforms/karaf/features/src/main/resources/features.xml
+++ b/platforms/karaf/features/src/main/resources/features.xml
@@ -702,6 +702,11 @@
<bundle dependency='true'>mvn:org.apache.servicemix.bundles/org.apache.servicemix.bundles.ibatis-sqlmap/${ibatis-bundle-version}</bundle>
<bundle>mvn:org.apache.camel/camel-ibatis/${project.version}</bundle>
</feature>
+ <feature name='camel-ignite' version='${project.version}' resolver='(obr)' start-level='50'>
+ <feature version='${project.version}'>camel-core</feature>
+ <feature version='${ignite-version-range}'>ignite-core</feature>
+ <bundle>mvn:org.apache.camel/camel-ignite/${project.version}</bundle>
+ </feature>
<feature name='camel-ical' version='${project.version}' resolver='(obr)' start-level='50'>
<bundle dependency='true'>mvn:commons-lang/commons-lang/${commons-lang-version}</bundle>
<bundle dependency='true'>mvn:commons-codec/commons-codec/${commons-codec-version}</bundle>
[3/6] camel git commit: CAMEL-9445 camel-ignite: New Apache Ignite
component. Initial work.
Posted by ra...@apache.org.
CAMEL-9445 camel-ignite: New Apache Ignite component. Initial work.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a695c5d3
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a695c5d3
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a695c5d3
Branch: refs/heads/master
Commit: a695c5d3656dcbac1e17b3109c7a6c340e8df7f0
Parents: ff8e1f1
Author: Raul Kripalani <ra...@apache.org>
Authored: Wed Dec 23 13:26:42 2015 +0000
Committer: Raul Kripalani <ra...@apache.org>
Committed: Wed Dec 23 15:18:57 2015 +0000
----------------------------------------------------------------------
components/camel-ignite/pom.xml | 92 ++++++
.../ignite/AbstractIgniteEndpoint.java | 99 ++++++
.../ignite/ClusterGroupExpression.java | 30 ++
.../ignite/ClusterGroupExpressions.java | 85 +++++
.../camel/component/ignite/IgniteComponent.java | 240 ++++++++++++++
.../camel/component/ignite/IgniteConstants.java | 59 ++++
.../camel/component/ignite/IgniteHelper.java | 38 +++
.../IgniteCacheContinuousQueryConsumer.java | 155 +++++++++
.../ignite/cache/IgniteCacheEndpoint.java | 311 +++++++++++++++++++
.../ignite/cache/IgniteCacheOperation.java | 26 ++
.../ignite/cache/IgniteCacheProducer.java | 208 +++++++++++++
.../ignite/compute/IgniteComputeEndpoint.java | 155 +++++++++
.../compute/IgniteComputeExecutionType.java | 26 ++
.../ignite/compute/IgniteComputeProducer.java | 275 ++++++++++++++++
.../ignite/events/IgniteEventsConsumer.java | 92 ++++++
.../ignite/events/IgniteEventsEndpoint.java | 146 +++++++++
.../ignite/idgen/IgniteIdGenEndpoint.java | 116 +++++++
.../ignite/idgen/IgniteIdGenOperation.java | 26 ++
.../ignite/idgen/IgniteIdGenProducer.java | 83 +++++
.../messaging/IgniteMessagingConsumer.java | 78 +++++
.../messaging/IgniteMessagingEndpoint.java | 123 ++++++++
.../messaging/IgniteMessagingProducer.java | 69 ++++
.../messaging/IgniteMessagingSendMode.java | 23 ++
.../ignite/queue/IgniteQueueEndpoint.java | 175 +++++++++++
.../ignite/queue/IgniteQueueOperation.java | 26 ++
.../ignite/queue/IgniteQueueProducer.java | 170 ++++++++++
.../component/ignite/set/IgniteSetEndpoint.java | 131 ++++++++
.../ignite/set/IgniteSetOperation.java | 26 ++
.../component/ignite/set/IgniteSetProducer.java | 120 +++++++
.../src/main/resources/META-INF/LICENSE.txt | 203 ++++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 +
.../services/org/apache/camel/component/ignite | 18 ++
.../component/ignite/AbstractIgniteTest.java | 63 ++++
.../ignite/IgniteCacheContinuousQueryTest.java | 281 +++++++++++++++++
.../camel/component/ignite/IgniteCacheTest.java | 237 ++++++++++++++
.../component/ignite/IgniteComputeTest.java | 192 ++++++++++++
.../component/ignite/IgniteEventsTest.java | 168 ++++++++++
.../camel/component/ignite/IgniteIdGenTest.java | 83 +++++
.../component/ignite/IgniteMessagingTest.java | 164 ++++++++++
.../camel/component/ignite/IgniteQueueTest.java | 257 +++++++++++++++
.../camel/component/ignite/IgniteSetTest.java | 176 +++++++++++
.../ignite/TestIgniteComputeResources.java | 149 +++++++++
.../src/test/resources/log4j.properties | 37 +++
components/pom.xml | 1 +
parent/pom.xml | 1 +
45 files changed, 5244 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-ignite/pom.xml b/components/camel-ignite/pom.xml
new file mode 100644
index 0000000..53af3f7
--- /dev/null
+++ b/components/camel-ignite/pom.xml
@@ -0,0 +1,92 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>2.17-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-ignite</artifactId>
+ <packaging>bundle</packaging>
+ <name>Camel :: Ignite</name>
+ <description>Camel Ignite component</description>
+
+ <properties>
+ <camel.osgi.export.pkg>org.apache.camel.component.ignite.*</camel.osgi.export.pkg>
+ <camel.osgi.export.service>org.apache.camel.spi.ComponentResolver;component=ignite</camel.osgi.export.service>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ </dependency>
+
+ <!-- Apache Ignite dependency -->
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-core</artifactId>
+ <version>${ignite-version}</version>
+ </dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-spring</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.google.truth</groupId>
+ <artifactId>truth</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.jayway.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+
+</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/AbstractIgniteEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/AbstractIgniteEndpoint.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/AbstractIgniteEndpoint.java
new file mode 100644
index 0000000..09791ca
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/AbstractIgniteEndpoint.java
@@ -0,0 +1,99 @@
+/**
+ * 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.ignite;
+
+import org.apache.camel.Component;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.ignite.Ignite;
+
+/**
+ * Base class for all Ignite endpoints.
+ */
+@UriEndpoint(scheme = "ignite:...", title = "Ignite endpoints", syntax = "ignite:...", label = "nosql,cache,compute", producerOnly = true)
+public abstract class AbstractIgniteEndpoint extends DefaultEndpoint {
+
+ protected IgniteComponent component;
+
+ @UriParam(defaultValue = "true")
+ private boolean propagateIncomingBodyIfNoReturnValue = true;
+
+ @UriParam
+ private boolean treatCollectionsAsCacheObjects;
+
+ public AbstractIgniteEndpoint(String endpointUri, Component component) {
+ super(endpointUri, component);
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return false;
+ }
+
+ protected IgniteComponent igniteComponent() {
+ if (component == null) {
+ component = (IgniteComponent) getComponent();
+ }
+ return component;
+ }
+
+ protected Ignite ignite() {
+ return igniteComponent().getIgnite();
+ }
+
+ /**
+ * Gets whether to propagate the incoming body if the return type of the underlying
+ * Ignite operation is void.
+ *
+ * @return
+ */
+ public boolean isPropagateIncomingBodyIfNoReturnValue() {
+ return propagateIncomingBodyIfNoReturnValue;
+ }
+
+ /**
+ * Sets whether to propagate the incoming body if the return type of the underlying
+ * Ignite operation is void.
+ *
+ * @param propagateIncomingBodyIfNoReturnValue
+ */
+ public void setPropagateIncomingBodyIfNoReturnValue(boolean propagateIncomingBodyIfNoReturnValue) {
+ this.propagateIncomingBodyIfNoReturnValue = propagateIncomingBodyIfNoReturnValue;
+ }
+
+ /**
+ * Gets whether to treat Collections as cache objects or as Collections of items to
+ * insert/update/compute, etc.
+ *
+ * @return
+ */
+ public boolean isTreatCollectionsAsCacheObjects() {
+ return treatCollectionsAsCacheObjects;
+ }
+
+ /**
+ * Sets whether to treat Collections as cache objects or as Collections of items to
+ * insert/update/compute, etc.
+ *
+ * @param treatCollectionsAsCacheObjects
+ */
+ public void setTreatCollectionsAsCacheObjects(boolean treatCollectionsAsCacheObjects) {
+ this.treatCollectionsAsCacheObjects = treatCollectionsAsCacheObjects;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/ClusterGroupExpression.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/ClusterGroupExpression.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/ClusterGroupExpression.java
new file mode 100644
index 0000000..b9286a1c
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/ClusterGroupExpression.java
@@ -0,0 +1,30 @@
+/**
+ * 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.ignite;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.cluster.ClusterGroup;
+
+/**
+ * Implement this interface to create an expression that returns a {@link ClusterGroup} from the
+ * {@link Ignite} instance that was passed in.
+ */
+public interface ClusterGroupExpression {
+
+ ClusterGroup getClusterGroup(Ignite ignite);
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/ClusterGroupExpressions.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/ClusterGroupExpressions.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/ClusterGroupExpressions.java
new file mode 100644
index 0000000..5ec660c
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/ClusterGroupExpressions.java
@@ -0,0 +1,85 @@
+/**
+ * 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.ignite;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.cluster.ClusterGroup;
+
+/**
+ * Convenient set of commonly used {@link ClusterGroupExpression}s.
+ */
+public final class ClusterGroupExpressions {
+
+ public static final ClusterGroupExpression FOR_CLIENTS = new ClusterGroupExpression() {
+ @Override
+ public ClusterGroup getClusterGroup(Ignite ignite) {
+ return ignite.cluster().forClients();
+ }
+ };
+
+ public static final ClusterGroupExpression FOR_DAEMONS = new ClusterGroupExpression() {
+ @Override
+ public ClusterGroup getClusterGroup(Ignite ignite) {
+ return ignite.cluster().forDaemons();
+ }
+ };
+
+ public static final ClusterGroupExpression FOR_LOCAL = new ClusterGroupExpression() {
+ @Override
+ public ClusterGroup getClusterGroup(Ignite ignite) {
+ return ignite.cluster().forLocal();
+ }
+ };
+
+ public static final ClusterGroupExpression FOR_OLDEST = new ClusterGroupExpression() {
+ @Override
+ public ClusterGroup getClusterGroup(Ignite ignite) {
+ return ignite.cluster().forOldest();
+ }
+ };
+
+ public static final ClusterGroupExpression FOR_YOUNGEST = new ClusterGroupExpression() {
+ @Override
+ public ClusterGroup getClusterGroup(Ignite ignite) {
+ return ignite.cluster().forYoungest();
+ }
+ };
+
+ public static final ClusterGroupExpression FOR_RANDOM = new ClusterGroupExpression() {
+ @Override
+ public ClusterGroup getClusterGroup(Ignite ignite) {
+ return ignite.cluster().forRandom();
+ }
+ };
+
+ public static final ClusterGroupExpression FOR_REMOTES = new ClusterGroupExpression() {
+ @Override
+ public ClusterGroup getClusterGroup(Ignite ignite) {
+ return ignite.cluster().forRemotes();
+ }
+ };
+
+ public static final ClusterGroupExpression FOR_SERVERS = new ClusterGroupExpression() {
+ @Override
+ public ClusterGroup getClusterGroup(Ignite ignite) {
+ return ignite.cluster().forServers();
+ }
+ };
+
+ private ClusterGroupExpressions() { }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteComponent.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteComponent.java
new file mode 100644
index 0000000..587d9be
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteComponent.java
@@ -0,0 +1,240 @@
+/**
+ * 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.ignite;
+
+import java.io.InputStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URL;
+import java.util.Map;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.ignite.cache.IgniteCacheEndpoint;
+import org.apache.camel.component.ignite.compute.IgniteComputeEndpoint;
+import org.apache.camel.component.ignite.events.IgniteEventsEndpoint;
+import org.apache.camel.component.ignite.idgen.IgniteIdGenEndpoint;
+import org.apache.camel.component.ignite.messaging.IgniteMessagingEndpoint;
+import org.apache.camel.component.ignite.queue.IgniteQueueEndpoint;
+import org.apache.camel.component.ignite.set.IgniteSetEndpoint;
+import org.apache.camel.impl.UriEndpointComponent;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.URISupport;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.configuration.IgniteConfiguration;
+
+/**
+ * The Ignite Component integrates Apache Camel with Apache Ignite, providing endpoints for the following functions:
+ * <ul>
+ * <li>Cache operations.</li>
+ * <li>Cluster computation.</li>
+ * <li>Messaging.</li>
+ * <li>Eventing.</li>
+ * <li>Id Generation.</li>
+ * <li>Set operations.</li>
+ * <li>Queue operations.</li>
+ * </ul>
+ */
+public class IgniteComponent extends UriEndpointComponent {
+
+ /**
+ * Modes of managing the underlying {@link Ignite} instance.
+ */
+ public enum IgniteLifecycleMode {
+ USER_MANAGED, COMPONENT_MANAGED
+ }
+
+ /** Ignite configuration. */
+ private IgniteConfiguration igniteConfiguration;
+
+ /** Resource from where to load configuration. */
+ private Object configurationResource;
+
+ /** Ignite instance. */
+ private Ignite ignite;
+
+ /** How the Ignite lifecycle is managed. */
+ private IgniteLifecycleMode lifecycleMode = IgniteLifecycleMode.COMPONENT_MANAGED;
+
+ public IgniteComponent() {
+ super(AbstractIgniteEndpoint.class);
+ }
+
+ public static IgniteComponent fromIgnite(Ignite ignite) {
+ IgniteComponent answer = new IgniteComponent();
+ answer.setIgnite(ignite);
+ return answer;
+ }
+
+ public static IgniteComponent fromConfiguration(IgniteConfiguration configuration) {
+ IgniteComponent answer = new IgniteComponent();
+ answer.setIgniteConfiguration(configuration);
+ return answer;
+ }
+
+ public static IgniteComponent fromInputStream(InputStream inputStream) {
+ IgniteComponent answer = new IgniteComponent();
+ answer.setConfigurationResource(inputStream);
+ return answer;
+ }
+
+ public static IgniteComponent fromUrl(URL url) {
+ IgniteComponent answer = new IgniteComponent();
+ answer.setConfigurationResource(url);
+ return answer;
+ }
+
+ public static IgniteComponent fromLocation(String location) {
+ IgniteComponent answer = new IgniteComponent();
+ answer.setConfigurationResource(location);
+ return answer;
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ ObjectHelper.notNull(getCamelContext(), "Camel Context");
+
+ AbstractIgniteEndpoint answer = null;
+ URI remainingUri = new URI(URISupport.normalizeUri(remaining));
+ String scheme = remainingUri.getScheme();
+
+ switch (scheme) {
+ case "cache":
+ answer = new IgniteCacheEndpoint(uri, remainingUri, parameters, this);
+ break;
+ case "compute":
+ answer = new IgniteComputeEndpoint(uri, remainingUri, parameters, this);
+ break;
+ case "messaging":
+ answer = new IgniteMessagingEndpoint(uri, remainingUri, parameters, this);
+ break;
+ case "events":
+ answer = new IgniteEventsEndpoint(uri, remainingUri, parameters, this);
+ break;
+ case "set":
+ answer = new IgniteSetEndpoint(uri, remainingUri, parameters, this);
+ break;
+ case "idgen":
+ answer = new IgniteIdGenEndpoint(uri, remainingUri, parameters, this);
+ break;
+ case "queue":
+ answer = new IgniteQueueEndpoint(uri, remainingUri, parameters, this);
+ break;
+
+ default:
+ throw new MalformedURLException("An invalid Ignite endpoint URI was provided. Please check that "
+ + "it starts with:" + " ignite:[cache/compute/messaging/...]:...");
+ }
+
+ setProperties(answer, parameters);
+
+ return answer;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ if (lifecycleMode == IgniteLifecycleMode.USER_MANAGED) {
+ return;
+ }
+
+ // Try to load the configuration from the resource.
+ if (configurationResource != null) {
+ if (configurationResource instanceof URL) {
+ ignite = Ignition.start((URL) configurationResource);
+ } else if (configurationResource instanceof InputStream) {
+ ignite = Ignition.start((InputStream) configurationResource);
+ } else if (configurationResource instanceof String) {
+ ignite = Ignition.start((String) configurationResource);
+ } else {
+ throw new IllegalStateException("An unsupported configuration resource was provided to the Ignite component. " + "Supported types are: URL, InputStream, String.");
+ }
+ } else if (igniteConfiguration != null) {
+ ignite = Ignition.start(igniteConfiguration);
+ } else {
+ throw new IllegalStateException("No configuration resource or IgniteConfiguration was provided to the Ignite component.");
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+
+ if (lifecycleMode == IgniteLifecycleMode.USER_MANAGED) {
+ return;
+ }
+
+ ignite.close();
+ }
+
+ /**
+ * Returns the {@link Ignite} instance.
+ *
+ * @return
+ */
+ public Ignite getIgnite() {
+ return ignite;
+ }
+
+ /**
+ * Sets the {@link Ignite} instance.
+ *
+ * @param ignite
+ */
+ public void setIgnite(Ignite ignite) {
+ this.ignite = ignite;
+ }
+
+ /**
+ * Gets the resource from where to load the configuration. It can be a: {@link URI}, {@link String} (URI)
+ * or an {@link InputStream}.
+ *
+ * @return
+ */
+ public Object getConfigurationResource() {
+ return configurationResource;
+ }
+
+ /**
+ * Sets the resource from where to load the configuration. It can be a: {@link URI}, {@link String} (URI)
+ * or an {@link InputStream}.
+ *
+ * @param configurationResource
+ */
+ public void setConfigurationResource(Object configurationResource) {
+ this.configurationResource = configurationResource;
+ }
+
+ /**
+ * Gets the {@link IgniteConfiguration} if the user set it explicitly.
+ * @return
+ */
+ public IgniteConfiguration getIgniteConfiguration() {
+ return igniteConfiguration;
+ }
+
+ /**
+ * Allows the user to set a programmatic {@link IgniteConfiguration}.
+ *
+ * @param igniteConfiguration
+ */
+ public void setIgniteConfiguration(IgniteConfiguration igniteConfiguration) {
+ this.igniteConfiguration = igniteConfiguration;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteConstants.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteConstants.java
new file mode 100644
index 0000000..e9b9256
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteConstants.java
@@ -0,0 +1,59 @@
+/**
+ * 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.ignite;
+
+/**
+ * Ignite Component constants.
+ */
+public final class IgniteConstants {
+
+ // Ignite Cache.
+ public static final String IGNITE_CACHE_KEY = "CamelIgniteCacheKey";
+ public static final String IGNITE_CACHE_QUERY = "CamelIgniteCacheQuery";
+ public static final String IGNITE_CACHE_OPERATION = "CamelIgniteCacheOperation";
+ public static final String IGNITE_CACHE_PEEK_MODE = "CamelIgniteCachePeekMode";
+ public static final String IGNITE_CACHE_EVENT_TYPE = "CamelIgniteCacheEventType";
+ public static final String IGNITE_CACHE_NAME = "CamelIgniteCacheName";
+ public static final String IGNITE_CACHE_OLD_VALUE = "CamelIgniteCacheOldValue";
+
+ // Ignite Messaging.
+ public static final String IGNITE_MESSAGING_TOPIC = "CamelIgniteMessagingTopic";
+ public static final String IGNITE_MESSAGING_UUID = "CamelIgniteMessagingUUID";
+
+ // Ignite Compute.
+ public static final String IGNITE_COMPUTE_EXECUTION_TYPE = "CamelIgniteComputeExecutionType";
+ public static final String IGNITE_COMPUTE_PARAMS = "CamelIgniteComputeParameters";
+ public static final String IGNITE_COMPUTE_REDUCER = "CamelIgniteComputeReducer";
+ public static final String IGNITE_COMPUTE_AFFINITY_CACHE_NAME = "CamelIgniteComputeAffinityCacheName";
+ public static final String IGNITE_COMPUTE_AFFINITY_KEY = "CamelIgniteComputeAffinityKey";
+
+ // Ignite Sets.
+ public static final String IGNITE_SETS_OPERATION = "CamelIgniteSetsOperation";
+ public static final String IGNITE_SETS_OPERATION_RESULT = "CamelIgniteSetsOperationResult";
+
+ // Ignite ID Gen.
+ public static final String IGNITE_IDGEN_OPERATION = "CamelIgniteIdGenOperation";
+
+ // Ignite Queues.
+ public static final String IGNITE_QUEUE_OPERATION = "CamelIgniteQueueOperation";
+ public static final String IGNITE_QUEUE_MAX_ELEMENTS = "CamelIgniteQueueMaxElements";
+ public static final String IGNITE_QUEUE_TRANSFERRED_COUNT = "CamelIgniteQueueTransferredCount";
+ public static final String IGNITE_QUEUE_TIMEOUT_MILLIS = "CamelIgniteQueueTimeoutMillis";
+
+ private IgniteConstants() { }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteHelper.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteHelper.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteHelper.java
new file mode 100644
index 0000000..05099f4
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/IgniteHelper.java
@@ -0,0 +1,38 @@
+/**
+ * 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.ignite;
+
+import org.apache.camel.Message;
+
+/**
+ * Helper methods for the Ignite component.
+ */
+public final class IgniteHelper {
+
+ private IgniteHelper() {
+
+ }
+
+ public static void maybePropagateIncomingBody(AbstractIgniteEndpoint endpoint, Message in, Message out) {
+ if (!endpoint.isPropagateIncomingBodyIfNoReturnValue()) {
+ return;
+ }
+
+ out.setBody(in.getBody());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java
new file mode 100644
index 0000000..e4c7302
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java
@@ -0,0 +1,155 @@
+/**
+ * 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.ignite.cache;
+
+import javax.cache.Cache.Entry;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.CacheEntryUpdatedListener;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.component.ignite.IgniteConstants;
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A consumer that generates {@link Exchange}s for items received from a continuous query.
+ */
+public class IgniteCacheContinuousQueryConsumer extends DefaultConsumer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IgniteCacheContinuousQueryConsumer.class);
+
+ private IgniteCacheEndpoint endpoint;
+
+ private IgniteCache<Object, Object> cache;
+
+ private QueryCursor<Entry<Object, Object>> cursor;
+
+ public IgniteCacheContinuousQueryConsumer(IgniteCacheEndpoint endpoint, Processor processor, IgniteCache<Object, Object> cache) {
+ super(endpoint, processor);
+ this.endpoint = endpoint;
+ this.cache = cache;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ launchContinuousQuery();
+
+ maybeFireExistingQueryResults();
+ }
+
+ private void maybeFireExistingQueryResults() {
+ if (!endpoint.isFireExistingQueryResults()) {
+ LOG.info(String.format("Skipping existing cache results for cache name = %s.", endpoint.getCacheName()));
+ return;
+ }
+
+ LOG.info(String.format("Processing existing cache results for cache name = %s.", endpoint.getCacheName()));
+
+ for (Entry<Object, Object> entry : cursor) {
+ Exchange exchange = createExchange(entry.getValue());
+ exchange.getIn().setHeader(IgniteConstants.IGNITE_CACHE_KEY, entry.getKey());
+ getAsyncProcessor().process(createExchange(entry), new AsyncCallback() {
+ @Override
+ public void done(boolean doneSync) {
+ // do nothing
+ }
+ });
+ }
+ }
+
+ private void launchContinuousQuery() {
+ ContinuousQuery<Object, Object> continuousQuery = new ContinuousQuery<>();
+
+ if (endpoint.getQuery() != null) {
+ continuousQuery.setInitialQuery(endpoint.getQuery());
+ }
+
+ if (endpoint.getRemoteFilter() != null) {
+ continuousQuery.setRemoteFilter(endpoint.getRemoteFilter());
+ }
+
+ continuousQuery.setLocalListener(new CacheEntryUpdatedListener<Object, Object>() {
+ @Override
+ public void onUpdated(Iterable<CacheEntryEvent<? extends Object, ? extends Object>> events) throws CacheEntryListenerException {
+ if (!endpoint.isOneExchangePerUpdate()) {
+ fireGroupedExchange(events);
+ return;
+ }
+
+ for (CacheEntryEvent<? extends Object, ? extends Object> entry : events) {
+ fireSingleExchange(entry);
+ }
+ }
+ });
+
+ continuousQuery.setAutoUnsubscribe(endpoint.isAutoUnsubscribe());
+ continuousQuery.setPageSize(endpoint.getPageSize());
+ continuousQuery.setTimeInterval(endpoint.getTimeInterval());
+
+ cursor = cache.query(continuousQuery);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+
+ cursor.close();
+ }
+
+ private void fireSingleExchange(CacheEntryEvent<? extends Object, ? extends Object> entry) {
+ Exchange exchange = createExchange(entry.getValue());
+ exchange.getIn().setHeader(IgniteConstants.IGNITE_CACHE_EVENT_TYPE, entry.getEventType());
+ exchange.getIn().setHeader(IgniteConstants.IGNITE_CACHE_OLD_VALUE, entry.getOldValue());
+ exchange.getIn().setHeader(IgniteConstants.IGNITE_CACHE_KEY, entry.getKey());
+ getAsyncProcessor().process(exchange, new AsyncCallback() {
+ @Override
+ public void done(boolean doneSync) {
+ // do nothing
+ }
+ });
+ }
+
+ private void fireGroupedExchange(Iterable<CacheEntryEvent<? extends Object, ? extends Object>> events) {
+ Exchange exchange = createExchange(events);
+ getAsyncProcessor().process(exchange, new AsyncCallback() {
+ @Override
+ public void done(boolean doneSync) {
+ // do nothing
+ }
+ });
+ }
+
+ private Exchange createExchange(Object payload) {
+ Exchange exchange = endpoint.createExchange(ExchangePattern.InOnly);
+ Message in = exchange.getIn();
+ in.setBody(payload);
+ in.setHeader(IgniteConstants.IGNITE_CACHE_NAME, endpoint.getCacheName());
+ return exchange;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheEndpoint.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheEndpoint.java
new file mode 100644
index 0000000..2b22409
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheEndpoint.java
@@ -0,0 +1,311 @@
+/**
+ * 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.ignite.cache;
+
+import java.net.URI;
+import java.util.Map;
+
+import javax.cache.Cache.Entry;
+
+import org.apache.camel.CamelException;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.ignite.AbstractIgniteEndpoint;
+import org.apache.camel.component.ignite.IgniteComponent;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.Query;
+
+/**
+ * Ignite Cache endpoint.
+ */
+@UriEndpoint(scheme = "ignite:cache", title = "Ignite Cache", syntax = "ignite:cache:[cacheName]", label = "nosql,cache,compute",
+ consumerClass = IgniteCacheContinuousQueryConsumer.class)
+public class IgniteCacheEndpoint extends AbstractIgniteEndpoint {
+
+ @UriParam
+ @Metadata(required = "true")
+ private String cacheName;
+
+ @UriParam
+ private IgniteCacheOperation operation;
+
+ @UriParam
+ private boolean failIfInexistentCache;
+
+ @UriParam
+ private CachePeekMode cachePeekMode = CachePeekMode.ALL;
+
+ @UriParam
+ private Query<Entry<Object, Object>> query;
+
+ @UriParam
+ private CacheEntryEventSerializableFilter<Object, Object> remoteFilter;
+
+ @UriParam
+ private boolean oneExchangePerUpdate = true;
+
+ @UriParam
+ private boolean fireExistingQueryResults;
+
+ @UriParam
+ private boolean autoUnsubscribe = ContinuousQuery.DFLT_AUTO_UNSUBSCRIBE;
+
+ @UriParam
+ private int pageSize = ContinuousQuery.DFLT_PAGE_SIZE;
+
+ @UriParam
+ private long timeInterval = ContinuousQuery.DFLT_TIME_INTERVAL;
+
+ public IgniteCacheEndpoint(String endpointUri, URI remainingUri, Map<String, Object> parameters, IgniteComponent igniteComponent) {
+ super(endpointUri, igniteComponent);
+ cacheName = remainingUri.getHost();
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new IgniteCacheProducer(this, obtainCache());
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ return new IgniteCacheContinuousQueryConsumer(this, processor, obtainCache());
+ }
+
+ private IgniteCache<Object, Object> obtainCache() throws CamelException {
+ IgniteCache<Object, Object> cache = ignite().cache(cacheName);
+ if (cache == null) {
+ if (failIfInexistentCache) {
+ throw new CamelException(String.format("Ignite cache %s doesn't exist, and failIfInexistentCache is true", cacheName));
+ }
+ cache = ignite().createCache(cacheName);
+ }
+
+ return cache;
+ }
+
+ /**
+ * Gets the cache name.
+ *
+ * @return
+ */
+ public String getCacheName() {
+ return cacheName;
+ }
+
+ /**
+ * Sets the cache name.
+ *
+ * @param cacheName
+ */
+ public void setCacheName(String cacheName) {
+ this.cacheName = cacheName;
+ }
+
+ /**
+ * Gets the cache operation to invoke.
+ *
+ * @return
+ */
+ public IgniteCacheOperation getOperation() {
+ return operation;
+ }
+
+ /**
+ * Sets the cache operation to invoke.
+ *
+ * @param operation
+ */
+ public void setOperation(IgniteCacheOperation operation) {
+ this.operation = operation;
+ }
+
+ /**
+ * Gets whether to fail the initialization if the cache doesn't exist.
+ *
+ * @return
+ */
+ public boolean isFailIfInexistentCache() {
+ return failIfInexistentCache;
+ }
+
+ /**
+ * Sets whether to fail the initialization if the cache doesn't exist.
+ *
+ * @param failIfInexistentCache
+ */
+ public void setFailIfInexistentCache(boolean failIfInexistentCache) {
+ this.failIfInexistentCache = failIfInexistentCache;
+ }
+
+ /**
+ * Gets the {@link CachePeekMode}, only needed for operations that require it ({@link IgniteCacheOperation#SIZE}).
+ *
+ * @return
+ */
+ public CachePeekMode getCachePeekMode() {
+ return cachePeekMode;
+ }
+
+ /**
+ * Sets the {@link CachePeekMode}, only needed for operations that require it ({@link IgniteCacheOperation#SIZE}).
+ *
+ * @param cachePeekMode
+ */
+ public void setCachePeekMode(CachePeekMode cachePeekMode) {
+ this.cachePeekMode = cachePeekMode;
+ }
+
+ /**
+ * Gets the query to execute, only needed for operations that require it,
+ * and for the Continuous Query Consumer.
+ *
+ * @return
+ */
+ public Query<Entry<Object, Object>> getQuery() {
+ return query;
+ }
+
+ /**
+ * Sets the query to execute, only needed for operations that require it,
+ * and for the Continuous Query Consumer.
+ *
+ * @param query
+ */
+ public void setQuery(Query<Entry<Object, Object>> query) {
+ this.query = query;
+ }
+
+ /**
+ * Gets the remote filter, only used by the Continuous Query Consumer.
+ *
+ * @return
+ */
+ public CacheEntryEventSerializableFilter<Object, Object> getRemoteFilter() {
+ return remoteFilter;
+ }
+
+ /**
+ * Sets the remote filter, only used by the Continuous Query Consumer.
+ *
+ * @param remoteFilter
+ */
+ public void setRemoteFilter(CacheEntryEventSerializableFilter<Object, Object> remoteFilter) {
+ this.remoteFilter = remoteFilter;
+ }
+
+ /**
+ * Gets whether to pack each update in an individual Exchange, even if multiple updates are
+ * received in one batch. Only used by the Continuous Query Consumer.
+ *
+ * @return
+ */
+ public boolean isOneExchangePerUpdate() {
+ return oneExchangePerUpdate;
+ }
+
+ /**
+ * Sets whether to pack each update in an individual Exchange, even if multiple updates are
+ * received in one batch. Only used by the Continuous Query Consumer.
+ *
+ * @param oneExchangePerUpdate
+ */
+ public void setOneExchangePerUpdate(boolean oneExchangePerUpdate) {
+ this.oneExchangePerUpdate = oneExchangePerUpdate;
+ }
+
+ /**
+ * Gets whether auto unsubscribe is enabled in the Continuous Query Consumer.
+ *
+ * @return
+ */
+ public boolean isAutoUnsubscribe() {
+ return autoUnsubscribe;
+ }
+
+ /**
+ * Sets whether auto unsubscribe is enabled in the Continuous Query Consumer.
+ *
+ * @param autoUnsubscribe
+ */
+ public void setAutoUnsubscribe(boolean autoUnsubscribe) {
+ this.autoUnsubscribe = autoUnsubscribe;
+ }
+
+ /**
+ * Gets the page size. Only used by the Continuous Query Consumer.
+ *
+ * @return
+ */
+ public int getPageSize() {
+ return pageSize;
+ }
+
+ /**
+ * Sets the page size. Only used by the Continuous Query Consumer.
+ *
+ * @param pageSize
+ */
+ public void setPageSize(int pageSize) {
+ this.pageSize = pageSize;
+ }
+
+ /**
+ * Gets whether to process existing results that match the query. Used on initialization of
+ * the Continuous Query Consumer.
+ *
+ * @return
+ */
+ public boolean isFireExistingQueryResults() {
+ return fireExistingQueryResults;
+ }
+
+ /**
+ * Sets whether to process existing results that match the query. Used on initialization of
+ * the Continuous Query Consumer.
+ *
+ * @param fireExistingQueryResults
+ */
+ public void setFireExistingQueryResults(boolean fireExistingQueryResults) {
+ this.fireExistingQueryResults = fireExistingQueryResults;
+ }
+
+ /**
+ * Gets the time interval for the Continuous Query Consumer.
+ *
+ * @return
+ */
+ public long getTimeInterval() {
+ return timeInterval;
+ }
+
+ /**
+ * Sets the time interval for the Continuous Query Consumer.
+ *
+ * @param timeInterval
+ */
+ public void setTimeInterval(long timeInterval) {
+ this.timeInterval = timeInterval;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheOperation.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheOperation.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheOperation.java
new file mode 100644
index 0000000..636df73
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheOperation.java
@@ -0,0 +1,26 @@
+/**
+ * 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.ignite.cache;
+
+/**
+ * Enumeration of Ignite Cache operations.
+ */
+public enum IgniteCacheOperation {
+
+ GET, PUT, REMOVE, SIZE, REBALANCE, QUERY, CLEAR
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheProducer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheProducer.java
new file mode 100644
index 0000000..7ad3f86
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheProducer.java
@@ -0,0 +1,208 @@
+/**
+ * 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.ignite.cache;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.ignite.IgniteConstants;
+import org.apache.camel.component.ignite.IgniteHelper;
+import org.apache.camel.impl.DefaultAsyncProducer;
+import org.apache.camel.spi.Synchronization;
+import org.apache.camel.util.MessageHelper;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.query.Query;
+import org.apache.ignite.cache.query.QueryCursor;
+
+/**
+ * Ignite Cache producer.
+ */
+public class IgniteCacheProducer extends DefaultAsyncProducer {
+
+ private IgniteCache<Object, Object> cache;
+ private IgniteCacheEndpoint endpoint;
+
+ public IgniteCacheProducer(IgniteCacheEndpoint endpoint, IgniteCache<Object, Object> igniteCache) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ this.cache = igniteCache;
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ Message in = exchange.getIn();
+ Message out = exchange.getOut();
+ MessageHelper.copyHeaders(exchange.getIn(), out, true);
+
+ switch (cacheOperationFor(exchange)) {
+
+ case GET:
+ doGet(in, out);
+ break;
+
+ case PUT:
+ doPut(in, out);
+ break;
+
+ case QUERY:
+ doQuery(in, out, exchange);
+ break;
+
+ case REMOVE:
+ doRemove(in, out);
+ break;
+
+ case CLEAR:
+ doClear(in, out);
+ break;
+
+ case SIZE:
+ doSize(in, out);
+ break;
+
+ case REBALANCE:
+ doRebalance(in, out);
+ break;
+
+ default:
+ break;
+ }
+
+ return true;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void doGet(Message in, Message out) {
+ Object cacheKey = cacheKey(in);
+
+ if (cacheKey instanceof Set && !endpoint.isTreatCollectionsAsCacheObjects()) {
+ out.setBody(cache.getAll((Set<Object>) cacheKey));
+ } else {
+ out.setBody(cache.get(cacheKey));
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void doPut(Message in, Message out) {
+ Map<Object, Object> map = in.getBody(Map.class);
+
+ if (map != null) {
+ cache.putAll(map);
+ return;
+ }
+
+ Object cacheKey = in.getHeader(IgniteConstants.IGNITE_CACHE_KEY);
+
+ if (cacheKey == null) {
+ throw new RuntimeCamelException("Cache PUT operation requires the cache key in the CamelIgniteCacheKey header, " + "or a payload of type Map.");
+ }
+
+ cache.put(cacheKey, in.getBody());
+
+ IgniteHelper.maybePropagateIncomingBody(endpoint, in, out);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void doQuery(Message in, Message out, Exchange exchange) {
+ Query<Object> query = in.getHeader(IgniteConstants.IGNITE_CACHE_QUERY, Query.class);
+
+ if (query == null) {
+ try {
+ query = in.getMandatoryBody(Query.class);
+ } catch (InvalidPayloadException e) {
+ exchange.setException(e);
+ return;
+ }
+ }
+
+ final QueryCursor<Object> cursor = cache.query(query);
+
+ out.setBody(cursor.iterator());
+
+ exchange.addOnCompletion(new Synchronization() {
+ @Override
+ public void onFailure(Exchange exchange) {
+ cursor.close();
+ }
+
+ @Override
+ public void onComplete(Exchange exchange) {
+ cursor.close();
+ }
+ });
+
+ }
+
+ @SuppressWarnings("unchecked")
+ private void doRemove(Message in, Message out) {
+ Object cacheKey = cacheKey(in);
+
+ if (cacheKey instanceof Set && !endpoint.isTreatCollectionsAsCacheObjects()) {
+ cache.removeAll((Set<Object>) cacheKey);
+ } else {
+ cache.remove(cacheKey);
+ }
+
+ IgniteHelper.maybePropagateIncomingBody(endpoint, in, out);
+ }
+
+ private void doClear(Message in, Message out) {
+ cache.removeAll();
+
+ IgniteHelper.maybePropagateIncomingBody(endpoint, in, out);
+ }
+
+ private void doRebalance(Message in, Message out) {
+ cache.rebalance().get();
+
+ IgniteHelper.maybePropagateIncomingBody(endpoint, in, out);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void doSize(Message in, Message out) {
+ Object peekMode = in.getHeader(IgniteConstants.IGNITE_CACHE_PEEK_MODE, endpoint.getCachePeekMode());
+
+ Integer result = null;
+ if (peekMode instanceof Collection) {
+ result = cache.size(((Collection<Object>) peekMode).toArray(new CachePeekMode[0]));
+ } else if (peekMode instanceof CachePeekMode) {
+ result = cache.size((CachePeekMode) peekMode);
+ }
+
+ out.setBody(result);
+ }
+
+ private Object cacheKey(Message msg) {
+ Object cacheKey = msg.getHeader(IgniteConstants.IGNITE_CACHE_KEY);
+ if (cacheKey == null) {
+ cacheKey = msg.getBody();
+ }
+ return cacheKey;
+ }
+
+ private IgniteCacheOperation cacheOperationFor(Exchange exchange) {
+ return exchange.getIn().getHeader(IgniteConstants.IGNITE_CACHE_OPERATION, endpoint.getOperation(), IgniteCacheOperation.class);
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java
new file mode 100644
index 0000000..1dc0663
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java
@@ -0,0 +1,155 @@
+/**
+ * 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.ignite.compute;
+
+import java.net.URI;
+import java.util.Map;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.ignite.AbstractIgniteEndpoint;
+import org.apache.camel.component.ignite.ClusterGroupExpression;
+import org.apache.camel.component.ignite.IgniteComponent;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCompute;
+
+/**
+ * Ignite Compute endpoint.
+ */
+@UriEndpoint(scheme = "ignite:compute", title = "Ignite Compute", syntax = "ignite:compute:endpointId", label = "nosql,cache,compute", producerOnly = true)
+public class IgniteComputeEndpoint extends AbstractIgniteEndpoint {
+
+ @UriParam
+ private ClusterGroupExpression clusterGroupExpression;
+
+ @UriParam
+ private IgniteComputeExecutionType executionType;
+
+ @UriParam
+ private String taskName;
+
+ @UriParam
+ private String computeName;
+
+ @UriParam
+ private Long timeoutMillis;
+
+ public IgniteComputeEndpoint(String uri, URI remainingUri, Map<String, Object> parameters, IgniteComponent igniteComponent) throws ClassNotFoundException {
+ super(uri, igniteComponent);
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new IgniteComputeProducer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ throw new UnsupportedOperationException("The Ignite Compute endpoint does not support consumers.");
+ }
+
+ public IgniteCompute createIgniteCompute() {
+ Ignite ignite = ignite();
+ IgniteCompute compute = clusterGroupExpression == null ? ignite.compute() : ignite.compute(clusterGroupExpression.getClusterGroup(ignite));
+
+ if (computeName != null) {
+ compute = compute.withName(computeName);
+ }
+
+ if (timeoutMillis != null) {
+ compute = compute.withTimeout(timeoutMillis);
+ }
+
+ return compute;
+ }
+
+ /**
+ * Gets the execution type of this producer.
+ *
+ * @return
+ */
+ public IgniteComputeExecutionType getExecutionType() {
+ return executionType;
+ }
+
+ /**
+ * Sets the execution type of this producer.
+ *
+ * @param executionType
+ */
+ public void setExecutionType(IgniteComputeExecutionType executionType) {
+ this.executionType = executionType;
+ }
+
+ /**
+ * Gets the task name, only applicable if using the {@link IgniteComputeExecutionType#EXECUTE} execution type.
+ *
+ * @return
+ */
+ public String getTaskName() {
+ return taskName;
+ }
+
+ /**
+ * Sets the task name, only applicable if using the {@link IgniteComputeExecutionType#EXECUTE} execution type.
+ *
+ * @param taskName
+ */
+ public void setTaskName(String taskName) {
+ this.taskName = taskName;
+ }
+
+ /**
+ * Gets the name of the compute job, which will be set via {@link IgniteCompute#withName(String)}.
+ *
+ * @return
+ */
+ public String getComputeName() {
+ return computeName;
+ }
+
+ /**
+ * Sets the name of the compute job, which will be set via {@link IgniteCompute#withName(String)}.
+ *
+ * @param computeName
+ */
+ public void setComputeName(String computeName) {
+ this.computeName = computeName;
+ }
+
+ /**
+ * Gets the timeout interval for triggered jobs, in milliseconds, which will be set via {@link IgniteCompute#withTimeout(long)}.
+ *
+ * @return
+ */
+ public Long getTimeoutMillis() {
+ return timeoutMillis;
+ }
+
+ /**
+ * Sets the timeout interval for triggered jobs, in milliseconds, which will be set via {@link IgniteCompute#withTimeout(long)}.
+ *
+ * @param timeoutMillis
+ */
+ public void setTimeoutMillis(Long timeoutMillis) {
+ this.timeoutMillis = timeoutMillis;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeExecutionType.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeExecutionType.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeExecutionType.java
new file mode 100644
index 0000000..f537a95
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeExecutionType.java
@@ -0,0 +1,26 @@
+/**
+ * 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.ignite.compute;
+
+/**
+ * Enumeration of execution type for the Ignite Compute producer.
+ */
+public enum IgniteComputeExecutionType {
+
+ CALL, BROADCAST, APPLY, EXECUTE, RUN, AFFINITY_CALL, AFFINITY_RUN
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeProducer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeProducer.java
new file mode 100644
index 0000000..d550a74
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeProducer.java
@@ -0,0 +1,275 @@
+/**
+ * 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.ignite.compute;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.TypeConverter;
+import org.apache.camel.component.ignite.IgniteConstants;
+import org.apache.camel.impl.DefaultAsyncProducer;
+import org.apache.camel.util.MessageHelper;
+import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.compute.ComputeTask;
+import org.apache.ignite.lang.IgniteCallable;
+import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.lang.IgniteReducer;
+import org.apache.ignite.lang.IgniteRunnable;
+
+/**
+ * Ignite Compute producer.
+ */
+public class IgniteComputeProducer extends DefaultAsyncProducer {
+
+ private IgniteComputeEndpoint endpoint;
+
+ public IgniteComputeProducer(IgniteComputeEndpoint endpoint) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ IgniteCompute compute = endpoint.createIgniteCompute().withAsync();
+
+ try {
+ switch (executionTypeFor(exchange)) {
+
+ case CALL:
+ doCall(exchange, callback, compute);
+ break;
+
+ case BROADCAST:
+ doBroadcast(exchange, callback, compute);
+ break;
+
+ case EXECUTE:
+ doExecute(exchange, callback, compute);
+ break;
+
+ case RUN:
+ doRun(exchange, callback, compute);
+ break;
+
+ case APPLY:
+ doApply(exchange, callback, compute);
+ break;
+
+ case AFFINITY_CALL:
+ doAffinityCall(exchange, callback, compute);
+ break;
+
+ case AFFINITY_RUN:
+ doAffinityRun(exchange, callback, compute);
+ break;
+
+ default:
+ exchange.setException(new UnsupportedOperationException("Operation not supported by Ignite Compute producer."));
+ return true;
+ }
+
+ compute.future().listen(IgniteInCamelClosure.create(exchange, callback));
+
+ } catch (Exception e) {
+ exchange.setException(e);
+ return true;
+ }
+
+ return false;
+ }
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ private void doCall(final Exchange exchange, final AsyncCallback callback, IgniteCompute compute) throws Exception {
+ Object job = exchange.getIn().getBody();
+ IgniteReducer<Object, Object> reducer = exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_REDUCER, IgniteReducer.class);
+
+ if (Collection.class.isAssignableFrom(job.getClass())) {
+ Collection<?> col = (Collection<?>) job;
+ TypeConverter tc = exchange.getContext().getTypeConverter();
+ Collection<IgniteCallable<?>> callables = new ArrayList<>(col.size());
+ for (Object o : col) {
+ callables.add(tc.mandatoryConvertTo(IgniteCallable.class, o));
+ }
+ if (reducer != null) {
+ compute.call((Collection) callables, reducer);
+ } else {
+ compute.call((Collection) callables);
+ }
+ } else if (IgniteCallable.class.isAssignableFrom(job.getClass())) {
+ compute.call((IgniteCallable<Object>) job);
+ } else {
+ throw new RuntimeCamelException(String.format(
+ "Ignite Compute endpoint with CALL executionType is only " + "supported for IgniteCallable payloads, or collections of them. The payload type was: %s.", job.getClass().getName()));
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void doBroadcast(final Exchange exchange, final AsyncCallback callback, IgniteCompute compute) throws Exception {
+ Object job = exchange.getIn().getBody();
+
+ if (IgniteCallable.class.isAssignableFrom(job.getClass())) {
+ compute.broadcast((IgniteCallable<?>) job);
+ } else if (IgniteRunnable.class.isAssignableFrom(job.getClass())) {
+ compute.broadcast((IgniteRunnable) job);
+ } else if (IgniteClosure.class.isAssignableFrom(job.getClass())) {
+ compute.broadcast((IgniteClosure<Object, Object>) job, exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_PARAMS));
+ } else {
+ throw new RuntimeCamelException(
+ String.format("Ignite Compute endpoint with BROADCAST executionType is only " + "supported for IgniteCallable, IgniteRunnable or IgniteClosure payloads. The payload type was: %s.",
+ job.getClass().getName()));
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void doExecute(final Exchange exchange, final AsyncCallback callback, IgniteCompute compute) throws Exception {
+ Object job = exchange.getIn().getBody();
+ Object params = exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_PARAMS);
+
+ if (job instanceof Class && ComputeTask.class.isAssignableFrom((Class<?>) job)) {
+ Class<? extends ComputeTask<Object, Object>> task = (Class<? extends ComputeTask<Object, Object>>) job;
+ compute.execute(task, params);
+ } else if (ComputeTask.class.isAssignableFrom(job.getClass())) {
+ compute.execute((ComputeTask<Object, Object>) job, params);
+ } else if (endpoint.getTaskName() != null) {
+ if (exchange.getIn().getBody() != null) {
+ params = exchange.getIn().getBody();
+ }
+ compute.execute(endpoint.getTaskName(), params);
+ } else {
+ throw new RuntimeCamelException(String.format("Ignite Compute endpoint with EXECUTE executionType is only "
+ + "supported for ComputeTask payloads, Class<ComputeTask> or any payload in conjunction with the " + "task name option. The payload type was: %s.", job.getClass().getName()));
+ }
+ }
+
+ private void doRun(final Exchange exchange, final AsyncCallback callback, IgniteCompute compute) throws Exception {
+ Object job = exchange.getIn().getBody();
+
+ if (Collection.class.isAssignableFrom(job.getClass())) {
+ Collection<?> col = (Collection<?>) job;
+ TypeConverter tc = exchange.getContext().getTypeConverter();
+ Collection<IgniteRunnable> runnables = new ArrayList<>(col.size());
+ for (Object o : col) {
+ runnables.add(tc.mandatoryConvertTo(IgniteRunnable.class, o));
+ }
+ compute.run(runnables);
+ } else if (IgniteRunnable.class.isAssignableFrom(job.getClass())) {
+ compute.run((IgniteRunnable) job);
+ } else {
+ throw new RuntimeCamelException(String.format(
+ "Ignite Compute endpoint with RUN executionType is only " + "supported for IgniteRunnable payloads, or collections of them. The payload type was: %s.", job.getClass().getName()));
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private <T, R1, R2> void doApply(final Exchange exchange, final AsyncCallback callback, IgniteCompute compute) throws Exception {
+ IgniteClosure<T, R1> job = exchange.getIn().getBody(IgniteClosure.class);
+ T params = (T) exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_PARAMS);
+
+ if (job == null || params == null) {
+ throw new RuntimeCamelException(
+ String.format("Ignite Compute endpoint with APPLY executionType is only " + "supported for IgniteClosure payloads with parameters. The payload type was: %s.",
+ exchange.getIn().getBody().getClass().getName()));
+ }
+
+ IgniteReducer<R1, R2> reducer = exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_REDUCER, IgniteReducer.class);
+
+ if (Collection.class.isAssignableFrom(params.getClass())) {
+ Collection<T> colParams = (Collection<T>) params;
+ if (reducer == null) {
+ compute.apply(job, colParams);
+ } else {
+ compute.apply(job, colParams, reducer);
+ }
+ } else {
+ compute.apply(job, params);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void doAffinityCall(final Exchange exchange, final AsyncCallback callback, IgniteCompute compute) throws Exception {
+ IgniteCallable<Object> job = exchange.getIn().getBody(IgniteCallable.class);
+ String affinityCache = exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_AFFINITY_CACHE_NAME, String.class);
+ Object affinityKey = exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_AFFINITY_KEY, Object.class);
+
+ if (job == null || affinityCache == null || affinityKey == null) {
+ throw new RuntimeCamelException(String.format(
+ "Ignite Compute endpoint with AFFINITY_CALL executionType is only " + "supported for IgniteCallable payloads, along with an affinity cache and key. The payload type was: %s.",
+ exchange.getIn().getBody().getClass().getName()));
+ }
+
+ compute.affinityCall(affinityCache, affinityKey, job);
+ }
+
+ private void doAffinityRun(final Exchange exchange, final AsyncCallback callback, IgniteCompute compute) throws Exception {
+ IgniteRunnable job = exchange.getIn().getBody(IgniteRunnable.class);
+ String affinityCache = exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_AFFINITY_CACHE_NAME, String.class);
+ Object affinityKey = exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_AFFINITY_KEY, Object.class);
+
+ if (job == null || affinityCache == null || affinityKey == null) {
+ throw new RuntimeCamelException(String.format(
+ "Ignite Compute endpoint with AFFINITY_RUN executionType is only " + "supported for IgniteRunnable payloads, along with an affinity cache and key. The payload type was: %s.",
+ exchange.getIn().getBody().getClass().getName()));
+ }
+
+ compute.affinityRun(affinityCache, affinityKey, job);
+ }
+
+ private IgniteComputeExecutionType executionTypeFor(Exchange exchange) {
+ return exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_EXECUTION_TYPE, endpoint.getExecutionType(), IgniteComputeExecutionType.class);
+ }
+
+ private static class IgniteInCamelClosure implements IgniteInClosure<IgniteFuture<Object>> {
+ private static final long serialVersionUID = 7486030906412223384L;
+
+ private Exchange exchange;
+ private AsyncCallback callback;
+
+ private static IgniteInCamelClosure create(Exchange exchange, AsyncCallback callback) {
+ IgniteInCamelClosure answer = new IgniteInCamelClosure();
+ answer.exchange = exchange;
+ answer.callback = callback;
+ return answer;
+ }
+
+ @Override
+ public void apply(IgniteFuture<Object> future) {
+ Message in = exchange.getIn();
+ Message out = exchange.getOut();
+ MessageHelper.copyHeaders(in, out, true);
+
+ Object result = null;
+
+ try {
+ result = future.get();
+ } catch (Exception e) {
+ exchange.setException(e);
+ callback.done(false);
+ return;
+ }
+
+ exchange.getOut().setBody(result);
+ callback.done(false);
+ }
+ };
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsConsumer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsConsumer.java
new file mode 100644
index 0000000..5d63611
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsConsumer.java
@@ -0,0 +1,92 @@
+/**
+ * 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.ignite.events;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.ignite.IgniteEvents;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Ignite Events consumer.
+ */
+public class IgniteEventsConsumer extends DefaultConsumer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IgniteEventsConsumer.class);
+
+ private IgniteEventsEndpoint endpoint;
+ private IgniteEvents events;
+ private int[] eventTypes = new int[0];
+
+ private IgnitePredicate<Event> predicate = new IgnitePredicate<Event>() {
+ private static final long serialVersionUID = 6738594728074592726L;
+
+ @Override
+ public boolean apply(Event event) {
+ Exchange exchange = endpoint.createExchange(ExchangePattern.InOnly);
+ Message in = exchange.getIn();
+ in.setBody(event);
+ try {
+ getAsyncProcessor().process(exchange, new AsyncCallback() {
+ @Override
+ public void done(boolean doneSync) {
+ // do nothing
+ }
+ });
+ } catch (Exception e) {
+ LOG.error(String.format("Exception while processing Ignite Event: %s.", event), e);
+ }
+ return true;
+ }
+ };
+
+ public IgniteEventsConsumer(IgniteEventsEndpoint endpoint, Processor processor, IgniteEvents events) {
+ super(endpoint, processor);
+ this.endpoint = endpoint;
+ this.events = events;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ if (endpoint.getEvents() != null && endpoint.getEvents().size() > 0) {
+ eventTypes = new int[endpoint.getEvents().size()];
+ int counter = 0;
+ for (Integer i : endpoint.getEvents()) {
+ eventTypes[counter++] = i;
+ }
+ }
+
+ events.localListen(predicate, eventTypes);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+
+ events.stopLocalListen(predicate, eventTypes);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsEndpoint.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsEndpoint.java
new file mode 100644
index 0000000..6237ad8
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsEndpoint.java
@@ -0,0 +1,146 @@
+/**
+ * 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.ignite.events;
+
+import java.lang.reflect.Field;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.ignite.AbstractIgniteEndpoint;
+import org.apache.camel.component.ignite.ClusterGroupExpression;
+import org.apache.camel.component.ignite.IgniteComponent;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteEvents;
+import org.apache.ignite.cluster.ClusterGroup;
+import org.apache.ignite.events.EventType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Ignite Events endpoint. Only supports consumers.
+ */
+@UriEndpoint(scheme = "ignite:events", title = "Ignite Events", syntax = "ignite:events:endpointId", label = "nosql,cache,compute,messaging,data",
+ consumerOnly = true, consumerClass = IgniteEventsConsumer.class)
+public class IgniteEventsEndpoint extends AbstractIgniteEndpoint {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IgniteEventsEndpoint.class);
+
+ @UriParam
+ private Set<Integer> events;
+
+ @UriParam
+ private ClusterGroupExpression clusterGroupExpression;
+
+ public IgniteEventsEndpoint(String uri, URI remainingUri, Map<String, Object> parameters, IgniteComponent igniteComponent) {
+ super(uri, igniteComponent);
+
+ // Initialize subscribed event types with ALL.
+ events = new HashSet<>();
+ for (Integer eventType : EventType.EVTS_ALL) {
+ events.add(eventType);
+ }
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ throw new UnsupportedOperationException("The Ignite Events endpoint does not support producers.");
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ // Initialize the Consumer.
+ IgniteEvents events = createIgniteEvents();
+ IgniteEventsConsumer consumer = new IgniteEventsConsumer(this, processor, events);
+ configureConsumer(consumer);
+
+ LOG.info("Created Ignite Events consumer for event types: {}.", events);
+
+ return consumer;
+ }
+
+ private IgniteEvents createIgniteEvents() {
+ Ignite ignite = ignite();
+ IgniteEvents events;
+ if (clusterGroupExpression == null) {
+ LOG.info("Ignite Events endpoint for event types {} using no Cluster Group.", this.events);
+ events = ignite.events();
+ } else {
+ ClusterGroup group = clusterGroupExpression.getClusterGroup(ignite);
+ LOG.info("Ignite Events endpoint for event types {} using Cluster Group: {}.", this.events, group);
+ events = ignite.events(group);
+ }
+ return events;
+ }
+
+ /**
+ * Gets the event types to subscribe to.
+ *
+ * @return
+ */
+ public Set<Integer> getEvents() {
+ return events;
+ }
+
+ /**
+ * Sets the event types to subscribe to as a {@link Set}.
+ *
+ * @param events
+ */
+ public void setEvents(Set<Integer> events) {
+ this.events = events;
+ }
+
+ /**
+ * Sets the event types to subscribe to as a comma-separated string of event constants as defined in {@link EventType}.
+ * <p>
+ * For example: EVT_CACHE_ENTRY_CREATED,EVT_CACHE_OBJECT_REMOVED,EVT_IGFS_DIR_CREATED.
+ *
+ * @param events
+ */
+ public void setEvents(String events) {
+ this.events = new HashSet<>();
+ Set<String> requestedEvents = new HashSet<>(Arrays.asList(events.toUpperCase().split(",")));
+ Field[] fields = EventType.class.getDeclaredFields();
+ for (Field field : fields) {
+ if (!requestedEvents.contains(field.getName())) {
+ continue;
+ }
+ try {
+ this.events.add(field.getInt(null));
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Problem while resolving event type. See stacktrace.", e);
+ }
+ }
+ }
+
+ public ClusterGroupExpression getClusterGroupExpression() {
+ return clusterGroupExpression;
+ }
+
+ public void setClusterGroupExpression(ClusterGroupExpression clusterGroupExpression) {
+ this.clusterGroupExpression = clusterGroupExpression;
+ }
+
+}
[2/6] camel git commit: CAMEL-9445 camel-ignite: New Apache Ignite
component. Initial work.
Posted by ra...@apache.org.
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenEndpoint.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenEndpoint.java
new file mode 100644
index 0000000..3e1da3c
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenEndpoint.java
@@ -0,0 +1,116 @@
+/**
+ * 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.ignite.idgen;
+
+import java.net.URI;
+import java.util.Map;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.ignite.AbstractIgniteEndpoint;
+import org.apache.camel.component.ignite.IgniteComponent;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.ignite.IgniteAtomicSequence;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Ignite ID Generator endpoint.
+ */
+@UriEndpoint(scheme = "ignite:idgen", title = "Ignite ID Generator", syntax = "ignite:idgen:[name]", label = "nosql,cache,compute", producerOnly = true)
+public class IgniteIdGenEndpoint extends AbstractIgniteEndpoint {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IgniteIdGenEndpoint.class);
+
+ @UriParam
+ @Metadata(required = "true")
+ private String name;
+
+ @UriParam
+ private Integer batchSize;
+
+ @UriParam(defaultValue = "0")
+ private Long initialValue = 0L;
+
+ @UriParam
+ private IgniteIdGenOperation operation;
+
+ public IgniteIdGenEndpoint(String endpointUri, URI remainingUri, Map<String, Object> parameters, IgniteComponent igniteComponent) throws Exception {
+ super(endpointUri, igniteComponent);
+ name = remainingUri.getHost();
+
+ ObjectHelper.notNull(name, "ID Generator name");
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ IgniteAtomicSequence atomicSeq = ignite().atomicSequence(name, initialValue, false);
+
+ if (atomicSeq == null) {
+ atomicSeq = ignite().atomicSequence(name, initialValue, true);
+ LOG.info("Created AtomicSequence of ID Generator with name {}.", name);
+ }
+
+ if (batchSize != null) {
+ atomicSeq.batchSize(batchSize);
+ }
+
+ return new IgniteIdGenProducer(this, atomicSeq);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ throw new UnsupportedOperationException("The Ignite Id Generator endpoint doesn't support consumers.");
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public Long getInitialValue() {
+ return initialValue;
+ }
+
+ public void setInitialValue(Long initialValue) {
+ this.initialValue = initialValue;
+ }
+
+ public IgniteIdGenOperation getOperation() {
+ return operation;
+ }
+
+ public void setOperation(IgniteIdGenOperation operation) {
+ this.operation = operation;
+ }
+
+ public Integer getBatchSize() {
+ return batchSize;
+ }
+
+ public void setBatchSize(Integer batchSize) {
+ this.batchSize = batchSize;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenOperation.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenOperation.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenOperation.java
new file mode 100644
index 0000000..d5d1db0
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenOperation.java
@@ -0,0 +1,26 @@
+/**
+ * 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.ignite.idgen;
+
+/**
+ * Enumeration of ID Generator / AtomicSequence operations.
+ */
+public enum IgniteIdGenOperation {
+
+ ADD_AND_GET, GET, GET_AND_ADD, GET_AND_INCREMENT, INCREMENT_AND_GET
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenProducer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenProducer.java
new file mode 100644
index 0000000..8ca75ae
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenProducer.java
@@ -0,0 +1,83 @@
+/**
+ * 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.ignite.idgen;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.component.ignite.IgniteConstants;
+import org.apache.camel.impl.DefaultAsyncProducer;
+import org.apache.camel.util.MessageHelper;
+import org.apache.ignite.IgniteAtomicSequence;
+
+/**
+ * Ignite ID Generator producer.
+ */
+public class IgniteIdGenProducer extends DefaultAsyncProducer {
+
+ private IgniteIdGenEndpoint endpoint;
+ private IgniteAtomicSequence atomicSeq;
+
+ public IgniteIdGenProducer(IgniteIdGenEndpoint endpoint, IgniteAtomicSequence atomicSeq) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ this.atomicSeq = atomicSeq;
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ Message in = exchange.getIn();
+ Message out = exchange.getOut();
+ MessageHelper.copyHeaders(in, out, true);
+
+ Long id = in.getBody(Long.class);
+
+ switch (idGenOperationFor(exchange)) {
+
+ case ADD_AND_GET:
+ out.setBody(atomicSeq.addAndGet(id));
+ break;
+
+ case GET:
+ out.setBody(atomicSeq.get());
+ break;
+
+ case GET_AND_ADD:
+ out.setBody(atomicSeq.getAndAdd(id));
+ break;
+
+ case GET_AND_INCREMENT:
+ out.setBody(atomicSeq.getAndIncrement());
+ break;
+
+ case INCREMENT_AND_GET:
+ out.setBody(atomicSeq.incrementAndGet());
+ break;
+
+ default:
+ exchange.setException(new UnsupportedOperationException("Operation not supported by Ignite ID Generator producer."));
+ return true;
+ }
+
+ return true;
+ }
+
+ private IgniteIdGenOperation idGenOperationFor(Exchange exchange) {
+ return exchange.getIn().getHeader(IgniteConstants.IGNITE_IDGEN_OPERATION, endpoint.getOperation(), IgniteIdGenOperation.class);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingConsumer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingConsumer.java
new file mode 100644
index 0000000..6579437
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingConsumer.java
@@ -0,0 +1,78 @@
+/**
+ * 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.ignite.messaging;
+
+import java.util.UUID;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.component.ignite.IgniteConstants;
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.ignite.IgniteMessaging;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class IgniteMessagingConsumer extends DefaultConsumer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IgniteMessagingConsumer.class);
+
+ private IgniteMessagingEndpoint endpoint;
+ private IgniteMessaging messaging;
+
+ private IgniteBiPredicate<UUID, Object> predicate = new IgniteBiPredicate<UUID, Object>() {
+ private static final long serialVersionUID = -971933058406324501L;
+
+ @Override
+ public boolean apply(UUID uuid, Object payload) {
+ Exchange exchange = endpoint.createExchange(ExchangePattern.InOnly);
+ Message in = exchange.getIn();
+ in.setBody(payload);
+ in.setHeader(IgniteConstants.IGNITE_MESSAGING_TOPIC, endpoint.getTopic());
+ in.setHeader(IgniteConstants.IGNITE_MESSAGING_UUID, uuid);
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ LOG.error(String.format("Exception while processing Ignite Message from topic %s", endpoint.getTopic()), e);
+ }
+ return true;
+ }
+ };
+
+ public IgniteMessagingConsumer(IgniteMessagingEndpoint endpoint, Processor processor, IgniteMessaging messaging) {
+ super(endpoint, processor);
+ this.endpoint = endpoint;
+ this.messaging = messaging;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ messaging.localListen(endpoint.getTopic(), predicate);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+
+ messaging.stopLocalListen(endpoint.getTopic(), predicate);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingEndpoint.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingEndpoint.java
new file mode 100644
index 0000000..2277a11
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingEndpoint.java
@@ -0,0 +1,123 @@
+/**
+ * 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.ignite.messaging;
+
+import java.net.URI;
+import java.util.Map;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.ignite.AbstractIgniteEndpoint;
+import org.apache.camel.component.ignite.ClusterGroupExpression;
+import org.apache.camel.component.ignite.IgniteComponent;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteMessaging;
+
+@UriEndpoint(scheme = "ignite:messaging", title = "Ignite Messaging", syntax = "ignite:messaging:[topic]", label = "nosql,cache,messaging")
+public class IgniteMessagingEndpoint extends AbstractIgniteEndpoint {
+
+ @UriParam
+ @Metadata(required = "true")
+ private String topic;
+
+ @UriParam
+ private ClusterGroupExpression clusterGroupExpression;
+
+ @UriParam
+ private IgniteMessagingSendMode sendMode = IgniteMessagingSendMode.UNORDERED;
+
+ @UriParam
+ private Long timeout;
+
+ public IgniteMessagingEndpoint(String endpointUri, URI remainingUri, Map<String, Object> parameters, IgniteComponent igniteComponent) {
+ super(endpointUri, igniteComponent);
+ topic = remainingUri.getHost();
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ // Validate options.
+ if (topic == null) {
+ throw new IllegalStateException("Cannot initialize an Ignite Messaging Producer with a null topic.");
+ }
+
+ if (sendMode == IgniteMessagingSendMode.ORDERED && timeout == null) {
+ throw new IllegalStateException("Cannot initialize an Ignite Messaging Producer in ORDERED send mode without a timeout.");
+ }
+
+ // Initialize the Producer.
+ IgniteMessaging messaging = createIgniteMessaging();
+ return new IgniteMessagingProducer(this, igniteComponent().getIgnite(), messaging);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ // Validate options.
+ if (topic == null) {
+ new IllegalStateException("Cannot initialize an Ignite Messaging Producer with a null topic.");
+ }
+
+ // Initialize the Consumer.
+ IgniteMessaging messaging = createIgniteMessaging();
+ IgniteMessagingConsumer consumer = new IgniteMessagingConsumer(this, processor, messaging);
+ configureConsumer(consumer);
+ return consumer;
+ }
+
+ private IgniteMessaging createIgniteMessaging() {
+ Ignite ignite = ignite();
+ IgniteMessaging messaging = clusterGroupExpression == null ? ignite.message() : ignite.message(clusterGroupExpression.getClusterGroup(ignite));
+ return messaging;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ public ClusterGroupExpression getClusterGroupExpression() {
+ return clusterGroupExpression;
+ }
+
+ public void setClusterGroupExpression(ClusterGroupExpression clusterGroupExpression) {
+ this.clusterGroupExpression = clusterGroupExpression;
+ }
+
+ public Long getTimeout() {
+ return timeout;
+ }
+
+ public void setTimeout(Long timeout) {
+ this.timeout = timeout;
+ }
+
+ public IgniteMessagingSendMode getSendMode() {
+ return sendMode;
+ }
+
+ public void setSendMode(IgniteMessagingSendMode sendMode) {
+ this.sendMode = sendMode;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingProducer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingProducer.java
new file mode 100644
index 0000000..c4946b8
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingProducer.java
@@ -0,0 +1,69 @@
+/**
+ * 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.ignite.messaging;
+
+import java.util.Collection;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.component.ignite.IgniteConstants;
+import org.apache.camel.component.ignite.IgniteHelper;
+import org.apache.camel.impl.DefaultAsyncProducer;
+import org.apache.camel.util.MessageHelper;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteMessaging;
+
+public class IgniteMessagingProducer extends DefaultAsyncProducer {
+
+ private IgniteMessagingEndpoint endpoint;
+ private IgniteMessaging messaging;
+
+ public IgniteMessagingProducer(IgniteMessagingEndpoint endpoint, Ignite ignite, IgniteMessaging messaging) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ this.messaging = messaging;
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ Message in = exchange.getIn();
+ Message out = exchange.getOut();
+ MessageHelper.copyHeaders(exchange.getIn(), out, true);
+
+ Object body = in.getBody();
+
+ if (endpoint.getSendMode() == IgniteMessagingSendMode.UNORDERED) {
+ if (body instanceof Collection<?> && !endpoint.isTreatCollectionsAsCacheObjects()) {
+ messaging.send(topicFor(exchange), (Collection<?>) body);
+ } else {
+ messaging.send(topicFor(exchange), body);
+ }
+ } else {
+ messaging.sendOrdered(topicFor(exchange), body, endpoint.getTimeout());
+ }
+
+ IgniteHelper.maybePropagateIncomingBody(endpoint, in, out);
+
+ return true;
+ }
+
+ private String topicFor(Exchange exchange) {
+ return exchange.getIn().getHeader(IgniteConstants.IGNITE_MESSAGING_TOPIC, endpoint.getTopic(), String.class);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingSendMode.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingSendMode.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingSendMode.java
new file mode 100644
index 0000000..0bf472c
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingSendMode.java
@@ -0,0 +1,23 @@
+/**
+ * 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.ignite.messaging;
+
+public enum IgniteMessagingSendMode {
+
+ ORDERED, UNORDERED
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueEndpoint.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueEndpoint.java
new file mode 100644
index 0000000..688a209
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueEndpoint.java
@@ -0,0 +1,175 @@
+/**
+ * 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.ignite.queue;
+
+import java.net.URI;
+import java.util.Map;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.ignite.AbstractIgniteEndpoint;
+import org.apache.camel.component.ignite.IgniteComponent;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.util.EndpointHelper;
+import org.apache.camel.util.IntrospectionSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.ignite.IgniteQueue;
+import org.apache.ignite.configuration.CollectionConfiguration;
+
+/**
+ * Ignite Queue endpoint.
+ */
+@UriEndpoint(scheme = "ignite:queue", title = "Ignite Queues", syntax = "ignite:queue:[name]", label = "nosql,cache", producerOnly = true)
+public class IgniteQueueEndpoint extends AbstractIgniteEndpoint {
+
+ @UriParam @Metadata(required = "true")
+ private String name;
+
+ @UriParam
+ private int capacity;
+
+ @UriParam
+ private CollectionConfiguration configuration = new CollectionConfiguration();
+
+ @UriParam
+ private Long timeoutMillis;
+
+ @UriParam
+ private IgniteQueueOperation operation;
+
+ public IgniteQueueEndpoint(String endpointUri, URI remainingUri, Map<String, Object> parameters, IgniteComponent igniteComponent) throws Exception {
+ super(endpointUri, igniteComponent);
+ name = remainingUri.getHost();
+
+ ObjectHelper.notNull(name, "Queue name");
+
+ // Set the configuration values.
+ if (!parameters.containsKey("configuration")) {
+ Map<String, Object> configProps = IntrospectionSupport.extractProperties(parameters, "config.");
+ EndpointHelper.setReferenceProperties(this.getCamelContext(), configProps, parameters);
+ EndpointHelper.setProperties(this.getCamelContext(), configProps, parameters);
+ }
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ IgniteQueue<Object> queue = ignite().queue(name, capacity, configuration);
+
+ return new IgniteQueueProducer(this, queue);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ throw new UnsupportedOperationException("The Ignite Queue endpoint doesn't support consumers.");
+ }
+
+ /**
+ * Gets the queue name.
+ *
+ * @return
+ */
+ public String getName() {
+ return name;
+ }
+
+ /**
+ * Sets the queue name.
+ *
+ * @param name
+ */
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ /**
+ * Gets the queue operation to perform.
+ *
+ * @return
+ */
+ public IgniteQueueOperation getOperation() {
+ return operation;
+ }
+
+ /**
+ * Sets the queue operation to perform.
+ *
+ * @param operation
+ */
+ public void setOperation(IgniteQueueOperation operation) {
+ this.operation = operation;
+ }
+
+ /**
+ * Gets the queue capacity. Default: non-bounded.
+ *
+ * @return
+ */
+ public int getCapacity() {
+ return capacity;
+ }
+
+ /**
+ * Sets the queue capacity. Default: non-bounded.
+ *
+ * @param capacity
+ */
+ public void setCapacity(int capacity) {
+ this.capacity = capacity;
+ }
+
+ /**
+ * Gets the collection configuration. Default: empty configuration.
+ *
+ * @return
+ */
+ public CollectionConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ /**
+ * Sets the collection configuration. Default: empty configuration.
+ * <p>
+ * You can also conveniently set inner properties by using <tt>configuration.xyz=123</tt> options.
+ *
+ * @param configuration
+ */
+ public void setConfiguration(CollectionConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ /**
+ * Gets the queue timeout in milliseconds. Default: no timeout.
+ *
+ * @return
+ */
+ public Long getTimeoutMillis() {
+ return timeoutMillis;
+ }
+
+ /**
+ * Sets the queue timeout in milliseconds. Default: no timeout.
+ *
+ * @param timeoutMillis
+ */
+ public void setTimeoutMillis(Long timeoutMillis) {
+ this.timeoutMillis = timeoutMillis;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueOperation.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueOperation.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueOperation.java
new file mode 100644
index 0000000..a342bcd
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueOperation.java
@@ -0,0 +1,26 @@
+/**
+ * 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.ignite.queue;
+
+/**
+ * Enumeration of queue operations.
+ */
+public enum IgniteQueueOperation {
+
+ CONTAINS, ADD, SIZE, REMOVE, ITERATOR, CLEAR, RETAIN_ALL, ARRAY, DRAIN, ELEMENT, PEEK, OFFER, POLL, TAKE, PUT
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueProducer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueProducer.java
new file mode 100644
index 0000000..191e62f
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueProducer.java
@@ -0,0 +1,170 @@
+/**
+ * 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.ignite.queue;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.component.ignite.IgniteConstants;
+import org.apache.camel.impl.DefaultAsyncProducer;
+import org.apache.camel.util.MessageHelper;
+import org.apache.ignite.IgniteQueue;
+
+public class IgniteQueueProducer extends DefaultAsyncProducer {
+
+ private IgniteQueueEndpoint endpoint;
+ private IgniteQueue<Object> queue;
+
+ public IgniteQueueProducer(IgniteQueueEndpoint endpoint, IgniteQueue<Object> queue) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ this.queue = queue;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ Message in = exchange.getIn();
+ Message out = exchange.getOut();
+ MessageHelper.copyHeaders(exchange.getIn(), out, true);
+
+ Object body = in.getBody();
+ Long millis;
+
+ switch (queueOperationFor(exchange)) {
+
+ case ADD:
+ if (Collection.class.isAssignableFrom(body.getClass()) && !endpoint.isTreatCollectionsAsCacheObjects()) {
+ out.setBody(queue.addAll((Collection<? extends Object>) body));
+ } else {
+ out.setBody(queue.add(body));
+ }
+ break;
+
+ case CONTAINS:
+ if (Collection.class.isAssignableFrom(body.getClass()) && !endpoint.isTreatCollectionsAsCacheObjects()) {
+ out.setBody(queue.containsAll((Collection<? extends Object>) body));
+ } else {
+ out.setBody(queue.contains(body));
+ }
+ break;
+
+ case SIZE:
+ out.setBody(queue.size());
+ break;
+
+ case REMOVE:
+ if (Collection.class.isAssignableFrom(body.getClass()) && !endpoint.isTreatCollectionsAsCacheObjects()) {
+ out.setBody(queue.removeAll((Collection<? extends Object>) body));
+ } else {
+ out.setBody(queue.remove(body));
+ }
+ break;
+
+ case CLEAR:
+ if (endpoint.isPropagateIncomingBodyIfNoReturnValue()) {
+ out.setBody(body);
+ }
+ queue.clear();
+ break;
+
+ case ITERATOR:
+ Iterator<?> iterator = queue.iterator();
+ out.setBody(iterator);
+ break;
+
+ case ARRAY:
+ out.setBody(queue.toArray());
+ break;
+
+ case RETAIN_ALL:
+ if (Collection.class.isAssignableFrom(body.getClass())) {
+ out.setBody(queue.retainAll((Collection<? extends Object>) body));
+ } else {
+ out.setBody(queue.retainAll(Collections.singleton(body)));
+ }
+ break;
+
+ case DRAIN:
+ Integer maxElements = in.getHeader(IgniteConstants.IGNITE_QUEUE_MAX_ELEMENTS, Integer.class);
+
+ Collection<Object> col = null;
+ if (body != null && Collection.class.isAssignableFrom(body.getClass())) {
+ col = (Collection<Object>) body;
+ } else {
+ col = maxElements != null ? new ArrayList<>(maxElements) : new ArrayList<>();
+ }
+
+ int transferred = -1;
+ if (maxElements == null) {
+ transferred = queue.drainTo(col);
+ } else {
+ transferred = queue.drainTo(col, maxElements);
+ }
+ out.setBody(col);
+ out.setHeader(IgniteConstants.IGNITE_QUEUE_TRANSFERRED_COUNT, transferred);
+ break;
+
+ case ELEMENT:
+ out.setBody(queue.element());
+ break;
+
+ case OFFER:
+ millis = in.getHeader(IgniteConstants.IGNITE_QUEUE_TIMEOUT_MILLIS, endpoint.getTimeoutMillis(), Long.class);
+ boolean result = millis == null ? queue.offer(body) : queue.offer(body, millis, TimeUnit.MILLISECONDS);
+ out.setBody(result);
+ break;
+
+ case PEEK:
+ out.setBody(queue.peek());
+ break;
+
+ case POLL:
+ millis = in.getHeader(IgniteConstants.IGNITE_QUEUE_TIMEOUT_MILLIS, endpoint.getTimeoutMillis(), Long.class);
+ out.setBody(millis == null ? queue.poll() : queue.poll(millis, TimeUnit.MILLISECONDS));
+ break;
+
+ case PUT:
+ if (endpoint.isPropagateIncomingBodyIfNoReturnValue()) {
+ out.setBody(in.getBody());
+ }
+ queue.put(body);
+ break;
+
+ case TAKE:
+ out.setBody(queue.take());
+ break;
+
+ default:
+ exchange.setException(new UnsupportedOperationException("Operation not supported by Ignite Queue producer."));
+ return true;
+ }
+
+ return true;
+ }
+
+ private IgniteQueueOperation queueOperationFor(Exchange exchange) {
+ return exchange.getIn().getHeader(IgniteConstants.IGNITE_QUEUE_OPERATION, endpoint.getOperation(), IgniteQueueOperation.class);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetEndpoint.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetEndpoint.java
new file mode 100644
index 0000000..6125ff9
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetEndpoint.java
@@ -0,0 +1,131 @@
+/**
+ * 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.ignite.set;
+
+import java.net.URI;
+import java.util.Map;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.ignite.AbstractIgniteEndpoint;
+import org.apache.camel.component.ignite.IgniteComponent;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.util.EndpointHelper;
+import org.apache.camel.util.IntrospectionSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.ignite.configuration.CollectionConfiguration;
+
+/**
+ * Ignite Cache endpoint.
+ */
+@UriEndpoint(scheme = "ignite:set", title = "Ignite Sets", syntax = "ignite:set:[name]", label = "nosql,cache", producerOnly = true)
+public class IgniteSetEndpoint extends AbstractIgniteEndpoint {
+
+ @UriParam @Metadata(required = "true")
+ private String name;
+
+ @UriParam
+ private CollectionConfiguration configuration = new CollectionConfiguration();
+
+ @UriParam
+ private IgniteSetOperation operation;
+
+ public IgniteSetEndpoint(String endpointUri, URI remainingUri, Map<String, Object> parameters, IgniteComponent igniteComponent) throws Exception {
+ super(endpointUri, igniteComponent);
+ name = remainingUri.getHost();
+
+ ObjectHelper.notNull(name, "Set name");
+
+ // Set the configuration values.
+ if (!parameters.containsKey("configuration")) {
+ Map<String, Object> configProps = IntrospectionSupport.extractProperties(parameters, "config.");
+ EndpointHelper.setReferenceProperties(this.getCamelContext(), configProps, parameters);
+ EndpointHelper.setProperties(this.getCamelContext(), configProps, parameters);
+ }
+
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new IgniteSetProducer(this, ignite().set(name, configuration));
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ throw new UnsupportedOperationException("The Ignite Sets endpoint doesn't support consumers.");
+ }
+
+ /**
+ * Gets the set name.
+ *
+ * @return
+ */
+ public String getName() {
+ return name;
+ }
+
+ /**
+ * Sets the set name.
+ *
+ * @param name
+ */
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ /**
+ * Gets the collection configuration. Default: empty configuration.
+ *
+ * @return
+ */
+ public CollectionConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ /**
+ * Sets the collection configuration. Default: empty configuration.
+ * <p>
+ * You can also conveniently set inner properties by using <tt>configuration.xyz=123</tt> options.
+ *
+ * @param configuration
+ */
+ public void setConfiguration(CollectionConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ /**
+ * Gets the set operation to perform.
+ *
+ * @return
+ */
+ public IgniteSetOperation getOperation() {
+ return operation;
+ }
+
+ /**
+ * Sets the set operation to perform.
+ *
+ * @param operation
+ */
+ public void setOperation(IgniteSetOperation operation) {
+ this.operation = operation;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetOperation.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetOperation.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetOperation.java
new file mode 100644
index 0000000..ad054e9
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetOperation.java
@@ -0,0 +1,26 @@
+/**
+ * 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.ignite.set;
+
+/**
+ * Enumeration of Set operations.
+ */
+public enum IgniteSetOperation {
+
+ CONTAINS, ADD, SIZE, REMOVE, ITERATOR, CLEAR, RETAIN_ALL, ARRAY
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetProducer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetProducer.java
new file mode 100644
index 0000000..918a0aa
--- /dev/null
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetProducer.java
@@ -0,0 +1,120 @@
+/**
+ * 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.ignite.set;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.component.ignite.IgniteConstants;
+import org.apache.camel.impl.DefaultAsyncProducer;
+import org.apache.camel.util.MessageHelper;
+import org.apache.ignite.IgniteSet;
+
+/**
+ * The Ignite Set producer.
+ */
+public class IgniteSetProducer extends DefaultAsyncProducer {
+
+ private IgniteSetEndpoint endpoint;
+ private IgniteSet<Object> set;
+
+ public IgniteSetProducer(IgniteSetEndpoint endpoint, IgniteSet<Object> set) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ this.set = set;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ Message in = exchange.getIn();
+ Message out = exchange.getOut();
+ MessageHelper.copyHeaders(exchange.getIn(), out, true);
+
+ Object body = in.getBody();
+
+ switch (setOperationFor(exchange)) {
+
+ case ADD:
+ if (Collection.class.isAssignableFrom(body.getClass()) && !endpoint.isTreatCollectionsAsCacheObjects()) {
+ out.setBody(set.addAll((Collection<? extends Object>) body));
+ } else {
+ out.setBody(set.add(body));
+ }
+ break;
+
+ case CONTAINS:
+ if (Collection.class.isAssignableFrom(body.getClass()) && !endpoint.isTreatCollectionsAsCacheObjects()) {
+ out.setBody(set.containsAll((Collection<? extends Object>) body));
+ } else {
+ out.setBody(set.contains(body));
+ }
+ break;
+
+ case SIZE:
+ out.setBody(set.size());
+ break;
+
+ case REMOVE:
+ if (Collection.class.isAssignableFrom(body.getClass()) && !endpoint.isTreatCollectionsAsCacheObjects()) {
+ out.setBody(set.removeAll((Collection<? extends Object>) body));
+ } else {
+ out.setBody(set.remove(body));
+ }
+ break;
+
+ case CLEAR:
+ if (endpoint.isPropagateIncomingBodyIfNoReturnValue()) {
+ out.setBody(body);
+ }
+ set.clear();
+ break;
+
+ case ITERATOR:
+ Iterator<?> iterator = set.iterator();
+ out.setBody(iterator);
+ break;
+
+ case ARRAY:
+ out.setBody(set.toArray());
+ break;
+
+ case RETAIN_ALL:
+ if (Collection.class.isAssignableFrom(body.getClass())) {
+ out.setBody(set.retainAll((Collection<? extends Object>) body));
+ } else {
+ out.setBody(set.retainAll(Collections.singleton(body)));
+ }
+ break;
+
+ default:
+ exchange.setException(new UnsupportedOperationException("Operation not supported by Ignite Set producer."));
+ return true;
+ }
+
+ return true;
+ }
+
+ private IgniteSetOperation setOperationFor(Exchange exchange) {
+ return exchange.getIn().getHeader(IgniteConstants.IGNITE_SETS_OPERATION, endpoint.getOperation(), IgniteSetOperation.class);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/resources/META-INF/LICENSE.txt
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/resources/META-INF/LICENSE.txt b/components/camel-ignite/src/main/resources/META-INF/LICENSE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/components/camel-ignite/src/main/resources/META-INF/LICENSE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed 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.
+
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/resources/META-INF/NOTICE.txt
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/resources/META-INF/NOTICE.txt b/components/camel-ignite/src/main/resources/META-INF/NOTICE.txt
new file mode 100644
index 0000000..2e215bf
--- /dev/null
+++ b/components/camel-ignite/src/main/resources/META-INF/NOTICE.txt
@@ -0,0 +1,11 @@
+ =========================================================================
+ == NOTICE file corresponding to the section 4 d of ==
+ == the Apache License, Version 2.0, ==
+ == in this case for the Apache Camel distribution. ==
+ =========================================================================
+
+ This product includes software developed by
+ The Apache Software Foundation (http://www.apache.org/).
+
+ Please read the different LICENSE files present in the licenses directory of
+ this distribution.
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/main/resources/META-INF/services/org/apache/camel/component/ignite
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/resources/META-INF/services/org/apache/camel/component/ignite b/components/camel-ignite/src/main/resources/META-INF/services/org/apache/camel/component/ignite
new file mode 100644
index 0000000..fca42c6
--- /dev/null
+++ b/components/camel-ignite/src/main/resources/META-INF/services/org/apache/camel/component/ignite
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+class=org.apache.camel.component.ignite.IgniteComponent
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/AbstractIgniteTest.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/AbstractIgniteTest.java b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/AbstractIgniteTest.java
new file mode 100644
index 0000000..d46cd1b
--- /dev/null
+++ b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/AbstractIgniteTest.java
@@ -0,0 +1,63 @@
+/**
+ * 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.ignite;
+
+import java.util.Collections;
+import java.util.UUID;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.EventType;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+
+public class AbstractIgniteTest extends CamelTestSupport {
+
+ /** Ip finder for TCP discovery. */
+ private static final TcpDiscoveryIpFinder LOCAL_IP_FINDER = new TcpDiscoveryVmIpFinder(false) { {
+ setAddresses(Collections.singleton("127.0.0.1:47500..47509"));
+ } };
+
+ private Ignite ignite;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.addComponent("ignite", buildComponent());
+ return context;
+ }
+
+ protected IgniteComponent buildComponent() {
+ IgniteConfiguration config = new IgniteConfiguration();
+ config.setGridName(UUID.randomUUID().toString());
+ config.setIncludeEventTypes(EventType.EVT_JOB_FINISHED, EventType.EVT_JOB_RESULTED);
+ config.setDiscoverySpi(new TcpDiscoverySpi().setIpFinder(LOCAL_IP_FINDER));
+
+ return IgniteComponent.fromConfiguration(config);
+ }
+
+ protected Ignite ignite() {
+ if (ignite == null) {
+ ignite = context.getComponent("ignite", IgniteComponent.class).getIgnite();
+ }
+ return ignite;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheContinuousQueryTest.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheContinuousQueryTest.java b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheContinuousQueryTest.java
new file mode 100644
index 0000000..c44b7f0
--- /dev/null
+++ b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheContinuousQueryTest.java
@@ -0,0 +1,281 @@
+/**
+ * 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.ignite;
+
+import java.io.Serializable;
+import java.util.Map;
+import java.util.Objects;
+
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.EventType;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Maps;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Route;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
+import org.apache.ignite.cache.query.ScanQuery;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.junit.After;
+import org.junit.Test;
+
+import static com.google.common.truth.Truth.assert_;
+
+public class IgniteCacheContinuousQueryTest extends AbstractIgniteTest implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ @Test
+ public void testContinuousQueryDoNotFireExistingEntries() throws Exception {
+ context.startRoute("continuousQuery");
+
+ getMockEndpoint("mock:test1").expectedMessageCount(100);
+
+ Map<Integer, Person> persons = createPersons(1, 100);
+ IgniteCache<Integer, Person> cache = ignite().getOrCreateCache("testcontinuous1");
+ cache.putAll(persons);
+
+ assertMockEndpointsSatisfied();
+
+ for (Exchange exchange : getMockEndpoint("mock:test1").getExchanges()) {
+ assert_().that(exchange.getIn().getHeader(IgniteConstants.IGNITE_CACHE_NAME)).isEqualTo("testcontinuous1");
+ assert_().that(exchange.getIn().getHeader(IgniteConstants.IGNITE_CACHE_EVENT_TYPE)).isEqualTo(EventType.CREATED);
+ assert_().that(exchange.getIn().getHeader(IgniteConstants.IGNITE_CACHE_KEY)).isIn(persons.keySet());
+ assert_().that(exchange.getIn().getBody()).isIn(persons.values());
+ }
+ }
+
+ @Test
+ public void testContinuousQueryFireExistingEntriesWithQuery() throws Exception {
+ getMockEndpoint("mock:test2").expectedMessageCount(50);
+
+ Map<Integer, Person> persons = createPersons(1, 100);
+ IgniteCache<Integer, Person> cache = ignite().getOrCreateCache("testcontinuous1");
+ cache.putAll(persons);
+
+ context.startRoute("continuousQuery.fireExistingEntries");
+
+ assertMockEndpointsSatisfied();
+
+ resetMocks();
+
+ getMockEndpoint("mock:test2").expectedMessageCount(100);
+
+ persons = createPersons(101, 100);
+ cache.putAll(persons);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testContinuousQueryFireExistingEntriesWithQueryAndRemoteFilter() throws Exception {
+ getMockEndpoint("mock:test3").expectedMessageCount(50);
+
+ Map<Integer, Person> persons = createPersons(1, 100);
+ IgniteCache<Integer, Person> cache = ignite().getOrCreateCache("testcontinuous1");
+ cache.putAll(persons);
+
+ context.startRoute("remoteFilter");
+
+ assertMockEndpointsSatisfied();
+
+ resetMocks();
+
+ getMockEndpoint("mock:test3").expectedMessageCount(50);
+
+ persons = createPersons(101, 100);
+ cache.putAll(persons);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testContinuousQueryGroupedUpdates() throws Exception {
+ // One hundred Iterables of 1 item each.
+ getMockEndpoint("mock:test4").expectedMessageCount(100);
+
+ context.startRoute("groupedUpdate");
+
+ Map<Integer, Person> persons = createPersons(1, 100);
+ IgniteCache<Integer, Person> cache = ignite().getOrCreateCache("testcontinuous1");
+ cache.putAll(persons);
+
+ assertMockEndpointsSatisfied();
+
+ for (Exchange exchange : getMockEndpoint("mock:test4").getExchanges()) {
+ assert_().that(exchange.getIn().getHeader(IgniteConstants.IGNITE_CACHE_NAME)).isEqualTo("testcontinuous1");
+ assert_().that(exchange.getIn().getBody()).isInstanceOf(Iterable.class);
+ assert_().that(Iterators.size(exchange.getIn().getBody(Iterable.class).iterator())).isEqualTo(1);
+ }
+
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("ignite:cache:testcontinuous1?query=#query1").routeId("continuousQuery").noAutoStartup().to("mock:test1");
+
+ from("ignite:cache:testcontinuous1?query=#query1&fireExistingQueryResults=true").routeId("continuousQuery.fireExistingEntries").noAutoStartup().to("mock:test2");
+
+ from("ignite:cache:testcontinuous1?query=#query1&remoteFilter=#remoteFilter1&fireExistingQueryResults=true").routeId("remoteFilter").noAutoStartup().to("mock:test3");
+
+ from("ignite:cache:testcontinuous1?pageSize=10&oneExchangePerUpdate=false").routeId("groupedUpdate").noAutoStartup().to("mock:test4");
+
+ }
+ };
+ }
+
+ private Map<Integer, Person> createPersons(int from, int count) {
+ Map<Integer, Person> answer = Maps.newHashMap();
+ int max = from + count;
+ for (int i = from; i < max; i++) {
+ answer.put(i, Person.create(i, "name" + i, "surname" + i));
+ }
+ return answer;
+ }
+
+ @Override
+ public boolean isCreateCamelContextPerClass() {
+ return true;
+ }
+
+ @After
+ public void deleteCaches() {
+ for (String cacheName : ImmutableSet.<String> of("testcontinuous1", "testcontinuous2", "testcontinuous3")) {
+ IgniteCache<?, ?> cache = ignite().cache(cacheName);
+ if (cache == null) {
+ continue;
+ }
+ cache.clear();
+ }
+ }
+
+ @After
+ public void stopAllRoutes() throws Exception {
+ for (Route route : context.getRoutes()) {
+ if (context.getRouteStatus(route.getId()) != ServiceStatus.Started) {
+ return;
+ }
+ context.stopRoute(route.getId());
+ }
+
+ resetMocks();
+ }
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry answer = super.createRegistry();
+
+ ScanQuery<Integer, Person> scanQuery1 = new ScanQuery<>(new IgniteBiPredicate<Integer, Person>() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean apply(Integer key, Person person) {
+ return person.getId() > 50;
+ }
+ });
+
+ CacheEntryEventSerializableFilter<Integer, Person> remoteFilter = new CacheEntryEventSerializableFilter<Integer, IgniteCacheContinuousQueryTest.Person>() {
+ private static final long serialVersionUID = 5624973479995548199L;
+
+ @Override
+ public boolean evaluate(CacheEntryEvent<? extends Integer, ? extends Person> event) throws CacheEntryListenerException {
+ return event.getValue().getId() > 150;
+ }
+ };
+
+ answer.bind("query1", scanQuery1);
+ answer.bind("remoteFilter1", remoteFilter);
+
+ return answer;
+ }
+
+ public static class Person implements Serializable {
+ private static final long serialVersionUID = -6582521698437964648L;
+
+ private Integer id;
+ private String name;
+ private String surname;
+
+ public static Person create(Integer id, String name, String surname) {
+ Person p = new Person();
+ p.setId(id);
+ p.setName(name);
+ p.setSurname(surname);
+ return p;
+ }
+
+ public Integer getId() {
+ return id;
+ }
+
+ public void setId(Integer id) {
+ this.id = id;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getSurname() {
+ return surname;
+ }
+
+ public void setSurname(String surname) {
+ this.surname = surname;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((id == null) ? 0 : id.hashCode());
+ result = prime * result + ((name == null) ? 0 : name.hashCode());
+ result = prime * result + ((surname == null) ? 0 : surname.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == null || !(obj instanceof Person)) {
+ return false;
+ }
+
+ if (this == obj) {
+ return true;
+ }
+
+ Person other = (Person) obj;
+ return Objects.equals(this.id, other.id) && Objects.equals(this.name, other.name) && Objects.equals(this.surname, other.surname);
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/a695c5d3/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheTest.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheTest.java b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheTest.java
new file mode 100644
index 0000000..b936e1b
--- /dev/null
+++ b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheTest.java
@@ -0,0 +1,237 @@
+/**
+ * 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.ignite;
+
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+
+import javax.cache.Cache.Entry;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Lists;
+
+import org.apache.camel.CamelException;
+import org.apache.camel.component.ignite.cache.IgniteCacheOperation;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.query.Query;
+import org.apache.ignite.cache.query.ScanQuery;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.junit.After;
+import org.junit.Test;
+
+import static com.google.common.truth.Truth.assert_;
+
+public class IgniteCacheTest extends AbstractIgniteTest {
+
+ @Test
+ public void testAddEntry() {
+ template.requestBodyAndHeader("ignite:cache:testcache1?operation=PUT", "1234", IgniteConstants.IGNITE_CACHE_KEY, "abcd");
+
+ assert_().that(ignite().cache("testcache1").size(CachePeekMode.ALL)).isEqualTo(1);
+ assert_().that(ignite().cache("testcache1").get("abcd")).isEqualTo("1234");
+ }
+
+ @Test
+ public void testAddEntrySet() {
+ template.requestBody("ignite:cache:testcache1?operation=PUT", ImmutableMap.of("abcd", "1234", "efgh", "5678"));
+
+ assert_().that(ignite().cache("testcache1").size(CachePeekMode.ALL)).isEqualTo(2);
+ assert_().that(ignite().cache("testcache1").get("abcd")).isEqualTo("1234");
+ assert_().that(ignite().cache("testcache1").get("efgh")).isEqualTo("5678");
+ }
+
+ @Test
+ public void testGetOne() {
+ testAddEntry();
+
+ String result = template.requestBody("ignite:cache:testcache1?operation=GET", "abcd", String.class);
+ assert_().that(result).isEqualTo("1234");
+
+ result = template.requestBodyAndHeader("ignite:cache:testcache1?operation=GET", "this value won't be used", IgniteConstants.IGNITE_CACHE_KEY, "abcd", String.class);
+ assert_().that(result).isEqualTo("1234");
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testGetMany() {
+ IgniteCache<String, String> cache = ignite().getOrCreateCache("testcache1");
+ Set<String> keys = new HashSet<>();
+
+ for (int i = 0; i < 100; i++) {
+ cache.put("k" + i, "v" + i);
+ keys.add("k" + i);
+ }
+
+ Map<String, String> result = template.requestBody("ignite:cache:testcache1?operation=GET", keys, Map.class);
+ for (String k : keys) {
+ assert_().that(result.get(k)).isEqualTo(k.replace("k", "v"));
+ }
+ }
+
+ @Test
+ public void testGetSize() {
+ IgniteCache<String, String> cache = ignite().getOrCreateCache("testcache1");
+ Set<String> keys = new HashSet<>();
+
+ for (int i = 0; i < 100; i++) {
+ cache.put("k" + i, "v" + i);
+ keys.add("k" + i);
+ }
+
+ Integer result = template.requestBody("ignite:cache:testcache1?operation=SIZE", keys, Integer.class);
+ assert_().that(result).isEqualTo(100);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testQuery() {
+ IgniteCache<String, String> cache = ignite().getOrCreateCache("testcache1");
+ Set<String> keys = new HashSet<>();
+
+ for (int i = 0; i < 100; i++) {
+ cache.put("k" + i, "v" + i);
+ keys.add("k" + i);
+ }
+
+ Query<Entry<String, String>> query = new ScanQuery<String, String>(new IgniteBiPredicate<String, String>() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean apply(String key, String value) {
+ return Integer.parseInt(key.replace("k", "")) >= 50;
+ }
+ });
+
+ Iterator<String> iter = template.requestBodyAndHeader("ignite:cache:testcache1?operation=QUERY", keys, IgniteConstants.IGNITE_CACHE_QUERY, query, Iterator.class);
+ ArrayList<Object> results = Lists.newArrayList(Iterators.toArray(iter, Object.class));
+ assert_().that(results.size()).isEqualTo(50);
+ }
+
+ @Test
+ public void testGetManyTreatCollectionsAsCacheObjects() {
+ IgniteCache<Object, String> cache = ignite().getOrCreateCache("testcache1");
+ Set<String> keys = new HashSet<>();
+
+ for (int i = 0; i < 100; i++) {
+ cache.put("k" + i, "v" + i);
+ keys.add("k" + i);
+ }
+
+ // Also add a cache entry with the entire Set as a key.
+ cache.put(keys, "---");
+
+ String result = template.requestBody("ignite:cache:testcache1?operation=GET&treatCollectionsAsCacheObjects=true", keys, String.class);
+ assert_().that(result).isEqualTo("---");
+ }
+
+ @Test
+ public void testRemoveEntry() {
+ IgniteCache<String, String> cache = ignite().getOrCreateCache("testcache1");
+
+ cache.put("abcd", "1234");
+ cache.put("efgh", "5678");
+
+ assert_().that(cache.size(CachePeekMode.ALL)).isEqualTo(2);
+
+ template.requestBody("ignite:cache:testcache1?operation=REMOVE", "abcd");
+
+ assert_().that(cache.size(CachePeekMode.ALL)).isEqualTo(1);
+ assert_().that(cache.get("abcd")).isNull();
+
+ template.requestBodyAndHeader("ignite:cache:testcache1?operation=REMOVE", "this value won't be used", IgniteConstants.IGNITE_CACHE_KEY, "efgh");
+
+ assert_().that(cache.size(CachePeekMode.ALL)).isEqualTo(0);
+ assert_().that(cache.get("efgh")).isNull();
+
+ }
+
+ @Test
+ public void testClearCache() {
+ IgniteCache<String, String> cache = ignite().getOrCreateCache("testcache1");
+ for (int i = 0; i < 100; i++) {
+ cache.put("k" + i, "v" + i);
+ }
+
+ assert_().that(cache.size(CachePeekMode.ALL)).isEqualTo(100);
+
+ template.requestBody("ignite:cache:testcache1?operation=CLEAR", "this value won't be used");
+
+ assert_().that(cache.size(CachePeekMode.ALL)).isEqualTo(0);
+ }
+
+ @Test
+ public void testHeaderSetRemoveEntry() {
+ testAddEntry();
+
+ String result = template.requestBody("ignite:cache:testcache1?operation=GET", "abcd", String.class);
+ assert_().that(result).isEqualTo("1234");
+
+ result = template.requestBodyAndHeader("ignite:cache:testcache1?operation=GET", "abcd", IgniteConstants.IGNITE_CACHE_OPERATION, IgniteCacheOperation.REMOVE, String.class);
+
+ // The body has not changed, but the cache entry is gone.
+ assert_().that(result).isEqualTo("abcd");
+ assert_().that(ignite().cache("testcache1").size(CachePeekMode.ALL)).isEqualTo(0);
+ }
+
+ @Test
+ public void testAddEntryNoCacheCreation() {
+ try {
+ template.requestBodyAndHeader("ignite:cache:testcache2?operation=PUT&failIfInexistentCache=true", "1234", IgniteConstants.IGNITE_CACHE_KEY, "abcd");
+ } catch (Exception e) {
+ assert_().that(ObjectHelper.getException(CamelException.class, e).getMessage()).startsWith("Ignite cache testcache2 doesn't exist");
+ return;
+ }
+
+ fail("Should have thrown an exception");
+ }
+
+ @Test
+ public void testAddEntryDoNotPropagateIncomingBody() {
+ Object result = template.requestBodyAndHeader("ignite:cache:testcache1?operation=PUT&propagateIncomingBodyIfNoReturnValue=false", "1234", IgniteConstants.IGNITE_CACHE_KEY, "abcd",
+ Object.class);
+
+ assert_().that(ignite().cache("testcache1").size(CachePeekMode.ALL)).isEqualTo(1);
+ assert_().that(ignite().cache("testcache1").get("abcd")).isEqualTo("1234");
+
+ assert_().that(result).isNull();
+ }
+
+ @Override
+ public boolean isCreateCamelContextPerClass() {
+ return true;
+ }
+
+ @After
+ public void deleteCaches() {
+ for (String cacheName : ImmutableSet.<String> of("testcache1", "testcache2")) {
+ IgniteCache<?, ?> cache = ignite().cache(cacheName);
+ if (cache == null) {
+ continue;
+ }
+ cache.clear();
+ }
+ }
+
+}
[5/6] camel git commit: CAMEL-9445 camel-ignite: Adjust endpoint
metadata and add logs to consumers.
Posted by ra...@apache.org.
CAMEL-9445 camel-ignite: Adjust endpoint metadata and add logs to consumers.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/99c55941
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/99c55941
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/99c55941
Branch: refs/heads/master
Commit: 99c559419847d1d635fbe4c6af909aea121996f2
Parents: d26af3d
Author: Raul Kripalani <ra...@apache.org>
Authored: Wed Dec 23 15:55:55 2015 +0000
Committer: Raul Kripalani <ra...@apache.org>
Committed: Wed Dec 23 15:55:55 2015 +0000
----------------------------------------------------------------------
.../ignite/cache/IgniteCacheContinuousQueryConsumer.java | 10 ++++++++++
.../component/ignite/compute/IgniteComputeEndpoint.java | 2 +-
.../component/ignite/events/IgniteEventsConsumer.java | 9 +++++++++
.../component/ignite/events/IgniteEventsEndpoint.java | 2 +-
.../ignite/messaging/IgniteMessagingConsumer.java | 10 ++++++++++
.../ignite/messaging/IgniteMessagingEndpoint.java | 6 +++++-
.../ignite/messaging/IgniteMessagingProducer.java | 3 +++
.../ignite/messaging/IgniteMessagingSendMode.java | 3 +++
8 files changed, 42 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/99c55941/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java
index e4c7302..cb06424 100644
--- a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.ignite.cache;
+import java.util.Arrays;
+
import javax.cache.Cache.Entry;
import javax.cache.event.CacheEntryEvent;
import javax.cache.event.CacheEntryListenerException;
@@ -59,6 +61,8 @@ public class IgniteCacheContinuousQueryConsumer extends DefaultConsumer {
launchContinuousQuery();
+ LOG.info("Started Ignite Cache Continuous Query consumer for cache {} with query: {}.", cache.getName(), endpoint.getQuery());
+
maybeFireExistingQueryResults();
}
@@ -96,6 +100,10 @@ public class IgniteCacheContinuousQueryConsumer extends DefaultConsumer {
continuousQuery.setLocalListener(new CacheEntryUpdatedListener<Object, Object>() {
@Override
public void onUpdated(Iterable<CacheEntryEvent<? extends Object, ? extends Object>> events) throws CacheEntryListenerException {
+ if (LOG.isTraceEnabled()) {
+ LOG.info("Processing Continuous Query event(s): {}.", events);
+ }
+
if (!endpoint.isOneExchangePerUpdate()) {
fireGroupedExchange(events);
return;
@@ -119,6 +127,8 @@ public class IgniteCacheContinuousQueryConsumer extends DefaultConsumer {
super.doStop();
cursor.close();
+
+ LOG.info("Stopped Ignite Cache Continuous Query consumer for cache {} with query: {}.", cache.getName(), endpoint.getQuery());
}
private void fireSingleExchange(CacheEntryEvent<? extends Object, ? extends Object> entry) {
http://git-wip-us.apache.org/repos/asf/camel/blob/99c55941/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java
index 1dc0663..d6a3eb2 100644
--- a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java
@@ -33,7 +33,7 @@ import org.apache.ignite.IgniteCompute;
/**
* Ignite Compute endpoint.
*/
-@UriEndpoint(scheme = "ignite:compute", title = "Ignite Compute", syntax = "ignite:compute:endpointId", label = "nosql,cache,compute", producerOnly = true)
+@UriEndpoint(scheme = "ignite:compute", title = "Ignite Compute", syntax = "ignite:compute:[endpointId]", label = "nosql,cache,compute", producerOnly = true)
public class IgniteComputeEndpoint extends AbstractIgniteEndpoint {
@UriParam
http://git-wip-us.apache.org/repos/asf/camel/blob/99c55941/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsConsumer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsConsumer.java
index 5d63611..7df33ac 100644
--- a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsConsumer.java
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsConsumer.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.ignite.events;
+import java.util.Arrays;
+
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
@@ -48,6 +50,9 @@ public class IgniteEventsConsumer extends DefaultConsumer {
Message in = exchange.getIn();
in.setBody(event);
try {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing Ignite Event: {}.", event);
+ }
getAsyncProcessor().process(exchange, new AsyncCallback() {
@Override
public void done(boolean doneSync) {
@@ -80,6 +85,8 @@ public class IgniteEventsConsumer extends DefaultConsumer {
}
events.localListen(predicate, eventTypes);
+
+ LOG.info("Started local Ignite Events consumer for events: {}.", Arrays.asList(eventTypes));
}
@Override
@@ -87,6 +94,8 @@ public class IgniteEventsConsumer extends DefaultConsumer {
super.doStop();
events.stopLocalListen(predicate, eventTypes);
+
+ LOG.info("Stopped local Ignite Events consumer for events: {}.", Arrays.asList(eventTypes));
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/99c55941/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsEndpoint.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsEndpoint.java
index 6237ad8..1d48a30 100644
--- a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsEndpoint.java
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/events/IgniteEventsEndpoint.java
@@ -41,7 +41,7 @@ import org.slf4j.LoggerFactory;
/**
* Ignite Events endpoint. Only supports consumers.
*/
-@UriEndpoint(scheme = "ignite:events", title = "Ignite Events", syntax = "ignite:events:endpointId", label = "nosql,cache,compute,messaging,data",
+@UriEndpoint(scheme = "ignite:events", title = "Ignite Events", syntax = "ignite:events:[endpointId]", label = "nosql,cache,compute,messaging,data",
consumerOnly = true, consumerClass = IgniteEventsConsumer.class)
public class IgniteEventsEndpoint extends AbstractIgniteEndpoint {
http://git-wip-us.apache.org/repos/asf/camel/blob/99c55941/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingConsumer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingConsumer.java
index 6579437..03a8c67 100644
--- a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingConsumer.java
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingConsumer.java
@@ -29,6 +29,9 @@ import org.apache.ignite.lang.IgniteBiPredicate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Ignite Messaging consumer.
+ */
public class IgniteMessagingConsumer extends DefaultConsumer {
private static final Logger LOG = LoggerFactory.getLogger(IgniteMessagingConsumer.class);
@@ -47,6 +50,9 @@ public class IgniteMessagingConsumer extends DefaultConsumer {
in.setHeader(IgniteConstants.IGNITE_MESSAGING_TOPIC, endpoint.getTopic());
in.setHeader(IgniteConstants.IGNITE_MESSAGING_UUID, uuid);
try {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing Ignite message for subscription {} with payload {}.", uuid, payload);
+ }
getProcessor().process(exchange);
} catch (Exception e) {
LOG.error(String.format("Exception while processing Ignite Message from topic %s", endpoint.getTopic()), e);
@@ -66,6 +72,8 @@ public class IgniteMessagingConsumer extends DefaultConsumer {
super.doStart();
messaging.localListen(endpoint.getTopic(), predicate);
+
+ LOG.info("Started Ignite Messaging consumer for topic {}.", endpoint.getTopic());
}
@Override
@@ -73,6 +81,8 @@ public class IgniteMessagingConsumer extends DefaultConsumer {
super.doStop();
messaging.stopLocalListen(endpoint.getTopic(), predicate);
+
+ LOG.info("Stopped Ignite Messaging consumer for topic {}.", endpoint.getTopic());
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/99c55941/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingEndpoint.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingEndpoint.java
index 2277a11..ca375f2 100644
--- a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingEndpoint.java
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingEndpoint.java
@@ -31,7 +31,11 @@ import org.apache.camel.spi.UriParam;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteMessaging;
-@UriEndpoint(scheme = "ignite:messaging", title = "Ignite Messaging", syntax = "ignite:messaging:[topic]", label = "nosql,cache,messaging")
+/**
+ * Ignite Messaging endpoint.
+ */
+@UriEndpoint(scheme = "ignite:messaging", title = "Ignite Messaging", syntax = "ignite:messaging:[topic]", label = "nosql,cache,messaging",
+ consumerClass = IgniteMessagingConsumer.class)
public class IgniteMessagingEndpoint extends AbstractIgniteEndpoint {
@UriParam
http://git-wip-us.apache.org/repos/asf/camel/blob/99c55941/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingProducer.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingProducer.java
index c4946b8..5a33c43 100644
--- a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingProducer.java
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingProducer.java
@@ -28,6 +28,9 @@ import org.apache.camel.util.MessageHelper;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteMessaging;
+/**
+ * Ignite Messaging producer.
+ */
public class IgniteMessagingProducer extends DefaultAsyncProducer {
private IgniteMessagingEndpoint endpoint;
http://git-wip-us.apache.org/repos/asf/camel/blob/99c55941/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingSendMode.java
----------------------------------------------------------------------
diff --git a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingSendMode.java b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingSendMode.java
index 0bf472c..a1c8900 100644
--- a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingSendMode.java
+++ b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/messaging/IgniteMessagingSendMode.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.component.ignite.messaging;
+/**
+ * Enum for Ignite Messaging send modes.
+ */
public enum IgniteMessagingSendMode {
ORDERED, UNORDERED
[6/6] camel git commit: CAMEL-9445 camel-ignite: New Apache Ignite
component. Merge code.
Posted by ra...@apache.org.
CAMEL-9445 camel-ignite: New Apache Ignite component. Merge code.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/0350423d
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/0350423d
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/0350423d
Branch: refs/heads/master
Commit: 0350423d99c6cc184bf21751c59a88ef5d3bb0fe
Parents: d20d0be 99c5594
Author: Raul Kripalani <ra...@apache.org>
Authored: Wed Dec 23 15:56:50 2015 +0000
Committer: Raul Kripalani <ra...@apache.org>
Committed: Wed Dec 23 15:56:50 2015 +0000
----------------------------------------------------------------------
components/camel-ignite/pom.xml | 92 ++++++
.../ignite/AbstractIgniteEndpoint.java | 99 ++++++
.../ignite/ClusterGroupExpression.java | 30 ++
.../ignite/ClusterGroupExpressions.java | 85 +++++
.../camel/component/ignite/IgniteComponent.java | 240 ++++++++++++++
.../camel/component/ignite/IgniteConstants.java | 59 ++++
.../camel/component/ignite/IgniteHelper.java | 38 +++
.../IgniteCacheContinuousQueryConsumer.java | 165 ++++++++++
.../ignite/cache/IgniteCacheEndpoint.java | 311 +++++++++++++++++++
.../ignite/cache/IgniteCacheOperation.java | 26 ++
.../ignite/cache/IgniteCacheProducer.java | 208 +++++++++++++
.../ignite/compute/IgniteComputeEndpoint.java | 155 +++++++++
.../compute/IgniteComputeExecutionType.java | 26 ++
.../ignite/compute/IgniteComputeProducer.java | 275 ++++++++++++++++
.../ignite/events/IgniteEventsConsumer.java | 101 ++++++
.../ignite/events/IgniteEventsEndpoint.java | 146 +++++++++
.../ignite/idgen/IgniteIdGenEndpoint.java | 116 +++++++
.../ignite/idgen/IgniteIdGenOperation.java | 26 ++
.../ignite/idgen/IgniteIdGenProducer.java | 83 +++++
.../messaging/IgniteMessagingConsumer.java | 88 ++++++
.../messaging/IgniteMessagingEndpoint.java | 127 ++++++++
.../messaging/IgniteMessagingProducer.java | 72 +++++
.../messaging/IgniteMessagingSendMode.java | 26 ++
.../ignite/queue/IgniteQueueEndpoint.java | 175 +++++++++++
.../ignite/queue/IgniteQueueOperation.java | 26 ++
.../ignite/queue/IgniteQueueProducer.java | 170 ++++++++++
.../component/ignite/set/IgniteSetEndpoint.java | 131 ++++++++
.../ignite/set/IgniteSetOperation.java | 26 ++
.../component/ignite/set/IgniteSetProducer.java | 120 +++++++
.../src/main/resources/META-INF/LICENSE.txt | 203 ++++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 +
.../services/org/apache/camel/component/ignite | 18 ++
.../component/ignite/AbstractIgniteTest.java | 63 ++++
.../ignite/IgniteCacheContinuousQueryTest.java | 281 +++++++++++++++++
.../camel/component/ignite/IgniteCacheTest.java | 237 ++++++++++++++
.../component/ignite/IgniteComputeTest.java | 192 ++++++++++++
.../component/ignite/IgniteEventsTest.java | 168 ++++++++++
.../camel/component/ignite/IgniteIdGenTest.java | 83 +++++
.../component/ignite/IgniteMessagingTest.java | 164 ++++++++++
.../camel/component/ignite/IgniteQueueTest.java | 257 +++++++++++++++
.../camel/component/ignite/IgniteSetTest.java | 176 +++++++++++
.../ignite/TestIgniteComputeResources.java | 149 +++++++++
.../src/test/resources/log4j.properties | 37 +++
components/pom.xml | 1 +
parent/pom.xml | 2 +
.../features/src/main/resources/features.xml | 5 +
46 files changed, 5289 insertions(+)
----------------------------------------------------------------------