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; + } + +}
