You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by lb...@apache.org on 2017/06/21 16:27:29 UTC
[4/6] camel git commit: CAMEL-10054: Create camel-atomix component
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/multimap/AtomixMultiMapProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/multimap/AtomixMultiMapProducer.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/multimap/AtomixMultiMapProducer.java
new file mode 100644
index 0000000..696b6eb
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/multimap/AtomixMultiMapProducer.java
@@ -0,0 +1,277 @@
+/**
+ * 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.atomix.client.multimap;
+
+import java.time.Duration;
+
+import io.atomix.collections.DistributedMultiMap;
+import io.atomix.resource.ReadConsistency;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.InvokeOnHeader;
+import org.apache.camel.Message;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientProducer;
+import org.apache.camel.util.ObjectHelper;
+
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_ACTION;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_KEY;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_NAME;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_READ_CONSISTENCY;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_TTL;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_VALUE;
+
+final class AtomixMultiMapProducer extends AbstractAtomixClientProducer<AtomixMultiMapEndpoint, DistributedMultiMap> {
+ private final AtomixMultiMapConfiguration configuration;
+
+ protected AtomixMultiMapProducer(AtomixMultiMapEndpoint endpoint) {
+ super(endpoint);
+ this.configuration = endpoint.getConfiguration();
+ }
+
+ // *********************************
+ // Handlers
+ // *********************************
+
+ @InvokeOnHeader("PUT")
+ boolean onPut(Message message, AsyncCallback callback) throws Exception {
+ final DistributedMultiMap<Object, Object> map = getResource(message);
+ final Object key = message.getHeader(RESOURCE_KEY, configuration::getKey, Object.class);
+ final Object val = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+ final long ttl = message.getHeader(RESOURCE_TTL, configuration::getTtl, long.class);
+
+ ObjectHelper.notNull(key, RESOURCE_KEY);
+ ObjectHelper.notNull(val, RESOURCE_VALUE);
+
+ if (ttl > 0) {
+ map.put(key, val, Duration.ofMillis(ttl)).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ map.put(key, val).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("GET")
+ boolean onGet(Message message, AsyncCallback callback) throws Exception {
+ final DistributedMultiMap<Object, Object> map = getResource(message);
+ final Object key = message.getHeader(RESOURCE_KEY, configuration::getKey, Object.class);
+ final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+
+ ObjectHelper.notNull(key, RESOURCE_KEY);
+
+ if (consistency != null) {
+ map.get(key, consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ map.get(key).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("CLEAR")
+ boolean onClear(Message message, AsyncCallback callback) throws Exception {
+ final DistributedMultiMap<Object, Object> map = getResource(message);
+
+ map.clear().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+
+ return false;
+ }
+
+ @InvokeOnHeader("SIZE")
+ boolean onSize(Message message, AsyncCallback callback) throws Exception {
+ final DistributedMultiMap<Object, Object> map = getResource(message);
+ final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+ final Object key = message.getHeader(RESOURCE_KEY, message::getBody, Object.class);
+
+ if (consistency != null) {
+ if (key != null) {
+ map.size(key, consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ map.size(consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+ } else {
+ if (key != null) {
+ map.size(key).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ map.size().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("IS_EMPTY")
+ boolean onIsEmpty(Message message, AsyncCallback callback) throws Exception {
+ final DistributedMultiMap<Object, Object> map = getResource(message);
+ final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+
+ if (consistency != null) {
+ map.isEmpty(consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ map.isEmpty().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("CONTAINS_KEY")
+ boolean onContainsKey(Message message, AsyncCallback callback) throws Exception {
+ final DistributedMultiMap<Object, Object> map = getResource(message);
+ final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+ final Object key = message.getHeader(RESOURCE_KEY, message::getBody, Object.class);
+
+ ObjectHelper.notNull(key, RESOURCE_KEY);
+
+ if (consistency != null) {
+ map.containsKey(key, consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ map.containsKey(key).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+// @InvokeOnHeader("CONTAINS_VALUE")
+// boolean onContainsValue(Message message, AsyncCallback callback) throws Exception {
+// final DistributedMultiMap<Object, Object> map = getResource(message);
+// final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+// final Object value = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+//
+// ObjectHelper.notNull(value, RESOURCE_VALUE);
+//
+// if (consistency != null) {
+// map.containsValue(value, consistency).thenAccept(
+// result -> processResult(message, callback, result)
+// );
+// } else {
+// map.containsValue(value).thenAccept(
+// result -> processResult(message, callback, result)
+// );
+// }
+//
+// return false;
+// }
+
+// @InvokeOnHeader("CONTAINS_ENTRY")
+// boolean onContainsEntry(Message message, AsyncCallback callback) throws Exception {
+// final DistributedMultiMap<Object, Object> map = getResource(message);
+// final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+// final Object key = message.getHeader(RESOURCE_KEY, message::getBody, Object.class);
+// final Object value = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+//
+// ObjectHelper.notNull(key, RESOURCE_VALUE);
+// ObjectHelper.notNull(value, RESOURCE_KEY);
+//
+// if (consistency != null) {
+// map.containsEntry(key, value, consistency).thenAccept(
+// result -> processResult(message, callback, result)
+// );
+// } else {
+// map.containsEntry(key, value).thenAccept(
+// result -> processResult(message, callback, result)
+// );
+// }
+//
+// return false;
+// }
+
+ @InvokeOnHeader("REMOVE")
+ boolean onRemove(Message message, AsyncCallback callback) throws Exception {
+ final DistributedMultiMap<Object, Object> map = getResource(message);
+ final Object key = message.getHeader(RESOURCE_KEY, message::getBody, Object.class);
+ final Object value = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+
+ ObjectHelper.notNull(key, RESOURCE_KEY);
+
+ if (value != null) {
+ map.remove(key, value).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ map.remove(key).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("REMOVE_VALUE")
+ boolean onRemoveValue(Message message, AsyncCallback callback) throws Exception {
+ final DistributedMultiMap<Object, Object> map = getResource(message);
+ final Object value = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+
+ ObjectHelper.notNull(value, RESOURCE_VALUE);
+
+ map.removeValue(value).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+
+ return false;
+ }
+
+ // *********************************
+ // Implementation
+ // *********************************
+
+
+ @Override
+ protected String getProcessorKey(Message message) {
+ return message.getHeader(RESOURCE_ACTION, configuration::getDefaultAction, String.class);
+ }
+
+ @Override
+ protected String getResourceName(Message message) {
+ return message.getHeader(RESOURCE_NAME, getAtomixEndpoint()::getResourceName, String.class);
+ }
+
+ @Override
+ protected DistributedMultiMap<Object, Object> createResource(String resourceName) {
+ return getAtomixEndpoint()
+ .getAtomix()
+ .getMultiMap(
+ resourceName,
+ new DistributedMultiMap.Config(getAtomixEndpoint().getConfiguration().getResourceOptions(resourceName)),
+ new DistributedMultiMap.Options(getAtomixEndpoint().getConfiguration().getResourceConfig(resourceName)))
+ .join();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueue.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueue.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueue.java
new file mode 100644
index 0000000..edc8cb5
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueue.java
@@ -0,0 +1,35 @@
+/**
+ * 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.atomix.client.queue;
+
+public final class AtomixQueue {
+
+ public enum Action {
+ ADD,
+ OFFER,
+ PEEK,
+ POLL,
+ CLEAR,
+ CONTAINS,
+ IS_EMPTY,
+ REMOVE,
+ SIZE
+ }
+
+ private AtomixQueue() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueComponent.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueComponent.java
new file mode 100644
index 0000000..083a851
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueComponent.java
@@ -0,0 +1,70 @@
+/**
+ * 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.atomix.client.queue;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientComponent;
+
+public final class AtomixQueueComponent extends AbstractAtomixClientComponent<AtomixQueueConfiguration> {
+ private AtomixQueueConfiguration configuration = new AtomixQueueConfiguration();
+
+ public AtomixQueueComponent() {
+ super();
+ }
+
+ public AtomixQueueComponent(CamelContext camelContext) {
+ super(camelContext);
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ AtomixQueueConfiguration configuration = this.configuration.copy();
+
+ // Bind options to the configuration object
+ setConfigurationProperties(configuration, parameters);
+
+ AtomixQueueEndpoint endpoint = new AtomixQueueEndpoint(uri, this, remaining);
+ endpoint.setConfiguration(configuration);
+
+ setProperties(endpoint, parameters);
+
+ return endpoint;
+ }
+
+ // **********************************************
+ // Properties
+ // **********************************************
+
+ public AtomixQueueConfiguration getConfiguration() {
+ return this.configuration;
+ }
+
+ /**
+ * The shared component configuration
+ */
+ public void setConfiguration(AtomixQueueConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ @Override
+ protected AtomixQueueConfiguration getComponentConfiguration() {
+ return getConfiguration();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueConfiguration.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueConfiguration.java
new file mode 100644
index 0000000..0175e69
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueConfiguration.java
@@ -0,0 +1,55 @@
+/**
+ * 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.atomix.client.queue;
+
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.atomix.client.AtomixClientConfiguration;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+
+@UriParams
+public class AtomixQueueConfiguration extends AtomixClientConfiguration {
+ @UriParam(defaultValue = "ADD")
+ private AtomixQueue.Action defaultAction = AtomixQueue.Action.ADD;
+
+ // ****************************************
+ // Properties
+ // ****************************************
+
+ public AtomixQueue.Action getDefaultAction() {
+ return defaultAction;
+ }
+
+ /**
+ * The default action.
+ */
+ public void setDefaultAction(AtomixQueue.Action defaultAction) {
+ this.defaultAction = defaultAction;
+ }
+
+ // ****************************************
+ // Copy
+ // ****************************************
+
+ public AtomixQueueConfiguration copy() {
+ try {
+ return (AtomixQueueConfiguration) super.clone();
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeCamelException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueConsumer.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueConsumer.java
new file mode 100644
index 0000000..eef6eec
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueConsumer.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.atomix.client.queue;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import io.atomix.catalyst.concurrent.Listener;
+import io.atomix.collections.DistributedQueue;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientConsumer;
+import org.apache.camel.component.atomix.client.AtomixClientConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class AtomixQueueConsumer extends AbstractAtomixClientConsumer<AtomixQueueEndpoint> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(AtomixQueueConsumer.class);
+
+ private final List<Listener<DistributedQueue.ValueEvent<Object>>> listeners;
+ private final String resourceName;
+ private final String resultHeader;
+ private DistributedQueue<Object> queue;
+
+ public AtomixQueueConsumer(AtomixQueueEndpoint endpoint, Processor processor, String resourceName) {
+ super(endpoint, processor);
+ this.listeners = new ArrayList<>();
+ this.resourceName = resourceName;
+ this.resultHeader = endpoint.getConfiguration().getResultHeader();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ this.queue = getAtomixEndpoint()
+ .getAtomix()
+ .getQueue(
+ resourceName,
+ new DistributedQueue.Config(getAtomixEndpoint().getConfiguration().getResourceOptions(resourceName)),
+ new DistributedQueue.Options(getAtomixEndpoint().getConfiguration().getResourceConfig(resourceName)))
+ .join();
+
+
+ LOGGER.debug("Subscribe to events for queue: {}", resourceName);
+ this.listeners.add(this.queue.onAdd(this::onEvent).join());
+ this.listeners.add(this.queue.onRemove(this::onEvent).join());
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ // close listeners
+ listeners.forEach(Listener::close);
+
+ super.doStart();
+ }
+
+ // ********************************************
+ // Event handler
+ // ********************************************
+
+ private void onEvent(DistributedQueue.ValueEvent<Object> event) {
+ Exchange exchange = getEndpoint().createExchange();
+ exchange.getIn().setHeader(AtomixClientConstants.EVENT_TYPE, event.type());
+
+ if (resultHeader == null) {
+ exchange.getIn().setBody(event.value());
+ } else {
+ exchange.getIn().setHeader(resultHeader, event.value());
+ }
+
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ getExceptionHandler().handleException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueEndpoint.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueEndpoint.java
new file mode 100644
index 0000000..4a16c1a
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueEndpoint.java
@@ -0,0 +1,61 @@
+/**
+ * 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.atomix.client.queue;
+
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientEndpoint;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+
+@UriEndpoint(
+ firstVersion = "2.20.0",
+ scheme = "atomix-queue",
+ title = "Atomix Queue",
+ syntax = "atomix-queue:queueName",
+ consumerClass = AtomixQueueConsumer.class,
+ label = "clustering")
+final class AtomixQueueEndpoint extends AbstractAtomixClientEndpoint<AtomixQueueComponent, AtomixQueueConfiguration> {
+ @UriParam
+ private AtomixQueueConfiguration configuration;
+
+ public AtomixQueueEndpoint(String uri, AtomixQueueComponent component, String resourceName) {
+ super(uri, component, resourceName);
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new AtomixQueueProducer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ return new AtomixQueueConsumer(this, processor, getResourceName());
+ }
+
+ @Override
+ public AtomixQueueConfiguration getConfiguration() {
+ return this.configuration;
+ }
+
+ @Override
+ public void setConfiguration(AtomixQueueConfiguration configuration) {
+ this.configuration = configuration;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueProducer.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueProducer.java
new file mode 100644
index 0000000..c207e34
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/queue/AtomixQueueProducer.java
@@ -0,0 +1,204 @@
+/**
+ * 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.atomix.client.queue;
+
+import io.atomix.collections.DistributedQueue;
+import io.atomix.resource.ReadConsistency;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.InvokeOnHeader;
+import org.apache.camel.Message;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientProducer;
+import org.apache.camel.util.ObjectHelper;
+
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_ACTION;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_NAME;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_READ_CONSISTENCY;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_VALUE;
+
+final class AtomixQueueProducer extends AbstractAtomixClientProducer<AtomixQueueEndpoint, DistributedQueue> {
+ private final AtomixQueueConfiguration configuration;
+
+ protected AtomixQueueProducer(AtomixQueueEndpoint endpoint) {
+ super(endpoint);
+ this.configuration = endpoint.getConfiguration();
+ }
+
+ // *********************************
+ // Handlers
+ // *********************************
+
+ @InvokeOnHeader("ADD")
+ boolean onAdd(Message message, AsyncCallback callback) throws Exception {
+ final DistributedQueue<Object> queue = getResource(message);
+ final Object val = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+
+ ObjectHelper.notNull(val, RESOURCE_VALUE);
+
+ queue.add(val).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+
+ return false;
+ }
+
+ @InvokeOnHeader("OFFER")
+ boolean onOffer(Message message, AsyncCallback callback) throws Exception {
+ final DistributedQueue<Object> queue = getResource(message);
+ final Object val = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+
+ ObjectHelper.notNull(val, RESOURCE_VALUE);
+
+ queue.offer(val).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+
+ return false;
+ }
+
+ @InvokeOnHeader("PEEK")
+ boolean onPeek(Message message, AsyncCallback callback) throws Exception {
+ final DistributedQueue<Object> queue = getResource(message);
+
+ queue.peek().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+
+ return false;
+ }
+
+ @InvokeOnHeader("POLL")
+ boolean onPoll(Message message, AsyncCallback callback) throws Exception {
+ final DistributedQueue<Object> queue = getResource(message);
+
+ queue.poll().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+
+ return false;
+ }
+
+ @InvokeOnHeader("CLEAR")
+ boolean onClear(Message message, AsyncCallback callback) throws Exception {
+ final DistributedQueue<Object> queue = getResource(message);
+
+ queue.clear().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+
+ return false;
+ }
+
+ @InvokeOnHeader("CONTAINS")
+ boolean onContains(Message message, AsyncCallback callback) throws Exception {
+ final DistributedQueue<Object> queue = getResource(message);
+ final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+ final Object value = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+
+ ObjectHelper.notNull(value, RESOURCE_VALUE);
+
+ if (consistency != null) {
+ queue.contains(value, consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ queue.contains(value).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("IS_EMPTY")
+ boolean onIsEmpty(Message message, AsyncCallback callback) throws Exception {
+ final DistributedQueue<Object> queue = getResource(message);
+ final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+
+ if (consistency != null) {
+ queue.isEmpty(consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ queue.isEmpty().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("REMOVE")
+ boolean onRemove(Message message, AsyncCallback callback) throws Exception {
+ final DistributedQueue<Object> queue = getResource(message);
+ final Object value = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+
+ if (value == null) {
+ queue.remove().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ queue.remove(value).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("SIZE")
+ boolean onSize(Message message, AsyncCallback callback) throws Exception {
+ final DistributedQueue<Object> queue = getResource(message);
+ final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+
+ if (consistency != null) {
+ queue.size(consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ queue.size().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ // *********************************
+ // Implementation
+ // *********************************
+
+ @Override
+ protected String getProcessorKey(Message message) {
+ return message.getHeader(RESOURCE_ACTION, configuration::getDefaultAction, String.class);
+ }
+
+ @Override
+ protected String getResourceName(Message message) {
+ return message.getHeader(RESOURCE_NAME, getAtomixEndpoint()::getResourceName, String.class);
+ }
+
+ @Override
+ protected DistributedQueue<Object> createResource(String resourceName) {
+ return getAtomixEndpoint()
+ .getAtomix()
+ .getQueue(
+ resourceName,
+ new DistributedQueue.Config(getAtomixEndpoint().getConfiguration().getResourceOptions(resourceName)),
+ new DistributedQueue.Options(getAtomixEndpoint().getConfiguration().getResourceConfig(resourceName)))
+ .join();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSet.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSet.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSet.java
new file mode 100644
index 0000000..ba5b24c
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSet.java
@@ -0,0 +1,32 @@
+/**
+ * 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.atomix.client.set;
+
+public final class AtomixSet {
+
+ public enum Action {
+ ADD,
+ CLEAR,
+ CONTAINS,
+ IS_EMPTY,
+ REMOVE,
+ SIZE
+ }
+
+ private AtomixSet() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetComponent.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetComponent.java
new file mode 100644
index 0000000..d91de4b
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetComponent.java
@@ -0,0 +1,70 @@
+/**
+ * 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.atomix.client.set;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientComponent;
+
+public final class AtomixSetComponent extends AbstractAtomixClientComponent<AtomixSetConfiguration> {
+ private AtomixSetConfiguration configuration = new AtomixSetConfiguration();
+
+ public AtomixSetComponent() {
+ super();
+ }
+
+ public AtomixSetComponent(CamelContext camelContext) {
+ super(camelContext);
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ AtomixSetConfiguration configuration = this.configuration.copy();
+
+ // Bind options to the configuration object
+ setConfigurationProperties(configuration, parameters);
+
+ AtomixSetEndpoint endpoint = new AtomixSetEndpoint(uri, this, remaining);
+ endpoint.setConfiguration(configuration);
+
+ setProperties(endpoint, parameters);
+
+ return endpoint;
+ }
+
+ // **********************************************
+ // Properties
+ // **********************************************
+
+ public AtomixSetConfiguration getConfiguration() {
+ return this.configuration;
+ }
+
+ /**
+ * The shared component configuration
+ */
+ public void setConfiguration(AtomixSetConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ @Override
+ protected AtomixSetConfiguration getComponentConfiguration() {
+ return getConfiguration();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetConfiguration.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetConfiguration.java
new file mode 100644
index 0000000..e7a67c9
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetConfiguration.java
@@ -0,0 +1,68 @@
+/**
+ * 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.atomix.client.set;
+
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.atomix.client.AtomixClientConfiguration;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+
+@UriParams
+public class AtomixSetConfiguration extends AtomixClientConfiguration {
+ @UriParam(defaultValue = "ADD")
+ private AtomixSet.Action defaultAction = AtomixSet.Action.ADD;
+ @UriParam
+ private long ttl;
+
+ // ****************************************
+ // Properties
+ // ****************************************
+
+ public AtomixSet.Action getDefaultAction() {
+ return defaultAction;
+ }
+
+ /**
+ * The default action.
+ */
+ public void setDefaultAction(AtomixSet.Action defaultAction) {
+ this.defaultAction = defaultAction;
+ }
+
+ public long getTtl() {
+ return ttl;
+ }
+
+ /**
+ * The resource ttl.
+ */
+ public void setTtl(long ttl) {
+ this.ttl = ttl;
+ }
+
+ // ****************************************
+ // Copy
+ // ****************************************
+
+ public AtomixSetConfiguration copy() {
+ try {
+ return (AtomixSetConfiguration) super.clone();
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeCamelException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetConsumer.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetConsumer.java
new file mode 100644
index 0000000..e20a719
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetConsumer.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.atomix.client.set;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import io.atomix.catalyst.concurrent.Listener;
+import io.atomix.collections.DistributedSet;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientConsumer;
+import org.apache.camel.component.atomix.client.AtomixClientConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class AtomixSetConsumer extends AbstractAtomixClientConsumer<AtomixSetEndpoint> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(AtomixSetConsumer.class);
+
+ private final List<Listener<DistributedSet.ValueEvent<Object>>> listeners;
+ private final String resourceName;
+ private final String resultHeader;
+ private DistributedSet<Object> set;
+
+ public AtomixSetConsumer(AtomixSetEndpoint endpoint, Processor processor, String resourceName) {
+ super(endpoint, processor);
+ this.listeners = new ArrayList<>();
+ this.resourceName = resourceName;
+ this.resultHeader = endpoint.getConfiguration().getResultHeader();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ this.set = getAtomixEndpoint()
+ .getAtomix()
+ .getSet(
+ resourceName,
+ new DistributedSet.Config(getAtomixEndpoint().getConfiguration().getResourceOptions(resourceName)),
+ new DistributedSet.Options(getAtomixEndpoint().getConfiguration().getResourceConfig(resourceName)))
+ .join();
+
+
+ LOGGER.debug("Subscribe to events for set: {}", resourceName);
+ this.listeners.add(this.set.onAdd(this::onEvent).join());
+ this.listeners.add(this.set.onRemove(this::onEvent).join());
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ // close listeners
+ listeners.forEach(Listener::close);
+
+ super.doStart();
+ }
+
+ // ********************************************
+ // Event handler
+ // ********************************************
+
+ private void onEvent(DistributedSet.ValueEvent<Object> event) {
+ Exchange exchange = getEndpoint().createExchange();
+ exchange.getIn().setHeader(AtomixClientConstants.EVENT_TYPE, event.type());
+
+ if (resultHeader == null) {
+ exchange.getIn().setBody(event.value());
+ } else {
+ exchange.getIn().setHeader(resultHeader, event.value());
+ }
+
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ getExceptionHandler().handleException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetEndpoint.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetEndpoint.java
new file mode 100644
index 0000000..9a1c64c
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetEndpoint.java
@@ -0,0 +1,61 @@
+/**
+ * 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.atomix.client.set;
+
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientEndpoint;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+
+@UriEndpoint(
+ firstVersion = "2.20.0",
+ scheme = "atomix-set",
+ title = "Atomix Set",
+ syntax = "atomix-set:setName",
+ consumerClass = AtomixSetConsumer.class,
+ label = "clustering")
+final class AtomixSetEndpoint extends AbstractAtomixClientEndpoint<AtomixSetComponent, AtomixSetConfiguration> {
+ @UriParam
+ private AtomixSetConfiguration configuration;
+
+ public AtomixSetEndpoint(String uri, AtomixSetComponent component, String resourceName) {
+ super(uri, component, resourceName);
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new AtomixSetProducer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ return new AtomixSetConsumer(this, processor, getResourceName());
+ }
+
+ @Override
+ public AtomixSetConfiguration getConfiguration() {
+ return this.configuration;
+ }
+
+ @Override
+ public void setConfiguration(AtomixSetConfiguration configuration) {
+ this.configuration = configuration;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetProducer.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetProducer.java
new file mode 100644
index 0000000..87a4406
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/set/AtomixSetProducer.java
@@ -0,0 +1,174 @@
+/**
+ * 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.atomix.client.set;
+
+import java.time.Duration;
+
+import io.atomix.collections.DistributedSet;
+import io.atomix.resource.ReadConsistency;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.InvokeOnHeader;
+import org.apache.camel.Message;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientProducer;
+import org.apache.camel.util.ObjectHelper;
+
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_ACTION;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_NAME;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_READ_CONSISTENCY;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_TTL;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_VALUE;
+
+final class AtomixSetProducer extends AbstractAtomixClientProducer<AtomixSetEndpoint, DistributedSet> {
+ private final AtomixSetConfiguration configuration;
+
+ protected AtomixSetProducer(AtomixSetEndpoint endpoint) {
+ super(endpoint);
+ this.configuration = endpoint.getConfiguration();
+ }
+
+ // *********************************
+ // Handlers
+ // *********************************
+
+ @InvokeOnHeader("ADD")
+ boolean onAdd(Message message, AsyncCallback callback) throws Exception {
+ final DistributedSet<Object> set = getResource(message);
+ final long ttl = message.getHeader(RESOURCE_TTL, configuration::getTtl, long.class);
+ final Object val = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+
+ ObjectHelper.notNull(val, RESOURCE_VALUE);
+
+ if (ttl > 0) {
+ set.add(val, Duration.ofMillis(ttl)).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ set.add(val).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("CLEAR")
+ boolean onClear(Message message, AsyncCallback callback) throws Exception {
+ final DistributedSet<Object> set = getResource(message);
+
+ set.clear().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+
+ return false;
+ }
+
+ @InvokeOnHeader("CONTAINS")
+ boolean onContains(Message message, AsyncCallback callback) throws Exception {
+ final DistributedSet<Object> set = getResource(message);
+ final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+ final Object value = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+
+ ObjectHelper.notNull(value, RESOURCE_VALUE);
+
+ if (consistency != null) {
+ set.contains(value, consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ set.contains(value).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("IS_EMPTY")
+ boolean onIsEmpty(Message message, AsyncCallback callback) throws Exception {
+ final DistributedSet<Object> set = getResource(message);
+ final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+
+ if (consistency != null) {
+ set.isEmpty(consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ set.isEmpty().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("REMOVE")
+ boolean onRemove(Message message, AsyncCallback callback) throws Exception {
+ final DistributedSet<Object> set = getResource(message);
+ final Object value = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+
+ ObjectHelper.notNull(value, RESOURCE_VALUE);
+
+ set.remove(value).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+
+ return false;
+ }
+
+ @InvokeOnHeader("SIZE")
+ boolean onSize(Message message, AsyncCallback callback) throws Exception {
+ final DistributedSet<Object> set = getResource(message);
+ final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+
+ if (consistency != null) {
+ set.size(consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ set.size().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ // *********************************
+ // Implementation
+ // *********************************
+
+ @Override
+ protected String getProcessorKey(Message message) {
+ return message.getHeader(RESOURCE_ACTION, configuration::getDefaultAction, String.class);
+ }
+
+ @Override
+ protected String getResourceName(Message message) {
+ return message.getHeader(RESOURCE_NAME, getAtomixEndpoint()::getResourceName, String.class);
+ }
+
+ @Override
+ protected DistributedSet<Object> createResource(String resourceName) {
+ return getAtomixEndpoint()
+ .getAtomix()
+ .getSet(
+ resourceName,
+ new DistributedSet.Config(getAtomixEndpoint().getConfiguration().getResourceOptions(resourceName)),
+ new DistributedSet.Options(getAtomixEndpoint().getConfiguration().getResourceConfig(resourceName)))
+ .join();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValue.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValue.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValue.java
new file mode 100644
index 0000000..0f082f6
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValue.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.atomix.client.value;
+
+public final class AtomixValue {
+
+ public enum Action {
+ SET,
+ GET,
+ GET_AND_SET,
+ COMPARE_AND_SET
+ }
+
+ private AtomixValue() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueComponent.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueComponent.java
new file mode 100644
index 0000000..d6d7ffb
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueComponent.java
@@ -0,0 +1,70 @@
+/**
+ * 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.atomix.client.value;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientComponent;
+
+public final class AtomixValueComponent extends AbstractAtomixClientComponent<AtomixValueConfiguration> {
+ private AtomixValueConfiguration configuration = new AtomixValueConfiguration();
+
+ public AtomixValueComponent() {
+ super();
+ }
+
+ public AtomixValueComponent(CamelContext camelContext) {
+ super(camelContext);
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ AtomixValueConfiguration configuration = this.configuration.copy();
+
+ // Bind options to the configuration object
+ setConfigurationProperties(configuration, parameters);
+
+ AtomixValueEndpoint endpoint = new AtomixValueEndpoint(uri, this, remaining);
+ endpoint.setConfiguration(configuration);
+
+ setProperties(endpoint, parameters);
+
+ return endpoint;
+ }
+
+ // **********************************************
+ // Properties
+ // **********************************************
+
+ public AtomixValueConfiguration getConfiguration() {
+ return this.configuration;
+ }
+
+ /**
+ * The shared component configuration
+ */
+ public void setConfiguration(AtomixValueConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ @Override
+ protected AtomixValueConfiguration getComponentConfiguration() {
+ return getConfiguration();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueConfiguration.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueConfiguration.java
new file mode 100644
index 0000000..79f8aa7
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueConfiguration.java
@@ -0,0 +1,68 @@
+/**
+ * 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.atomix.client.value;
+
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.atomix.client.AtomixClientConfiguration;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+
+@UriParams
+public class AtomixValueConfiguration extends AtomixClientConfiguration {
+ @UriParam(defaultValue = "SET")
+ private AtomixValue.Action defaultAction = AtomixValue.Action.SET;
+ @UriParam
+ private long ttl;
+
+ // ****************************************
+ // Properties
+ // ****************************************
+
+ public AtomixValue.Action getDefaultAction() {
+ return defaultAction;
+ }
+
+ /**
+ * The default action.
+ */
+ public void setDefaultAction(AtomixValue.Action defaultAction) {
+ this.defaultAction = defaultAction;
+ }
+
+ public long getTtl() {
+ return ttl;
+ }
+
+ /**
+ * The resource ttl.
+ */
+ public void setTtl(long ttl) {
+ this.ttl = ttl;
+ }
+
+ // ****************************************
+ // Copy
+ // ****************************************
+
+ public AtomixValueConfiguration copy() {
+ try {
+ return (AtomixValueConfiguration) super.clone();
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeCamelException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueConsumer.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueConsumer.java
new file mode 100644
index 0000000..ecb759e
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueConsumer.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.atomix.client.value;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import io.atomix.catalyst.concurrent.Listener;
+import io.atomix.variables.DistributedValue;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientConsumer;
+import org.apache.camel.component.atomix.client.AtomixClientConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class AtomixValueConsumer extends AbstractAtomixClientConsumer<AtomixValueEndpoint> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(AtomixValueConsumer.class);
+
+ private final List<Listener<DistributedValue.ChangeEvent<Object>>> listeners;
+ private final String resourceName;
+ private final String resultHeader;
+ private DistributedValue<Object> value;
+
+ public AtomixValueConsumer(AtomixValueEndpoint endpoint, Processor processor, String resourceName) {
+ super(endpoint, processor);
+ this.listeners = new ArrayList<>();
+ this.resourceName = resourceName;
+ this.resultHeader = endpoint.getConfiguration().getResultHeader();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ this.value = getAtomixEndpoint()
+ .getAtomix()
+ .getValue(
+ resourceName,
+ new DistributedValue.Config(getAtomixEndpoint().getConfiguration().getResourceOptions(resourceName)),
+ new DistributedValue.Options(getAtomixEndpoint().getConfiguration().getResourceConfig(resourceName)))
+ .join();
+
+
+ LOGGER.debug("Subscribe to events for queue: {}", resourceName);
+ this.listeners.add(this.value.onChange(this::onEvent).join());
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ // close listeners
+ listeners.forEach(Listener::close);
+
+ super.doStart();
+ }
+
+ // ********************************************
+ // Event handler
+ // ********************************************
+
+ private void onEvent(DistributedValue.ChangeEvent<Object> event) {
+ Exchange exchange = getEndpoint().createExchange();
+ exchange.getIn().setHeader(AtomixClientConstants.EVENT_TYPE, event.type());
+ exchange.getIn().setHeader(AtomixClientConstants.RESOURCE_OLD_VALUE, event.oldValue());
+
+ if (resultHeader == null) {
+ exchange.getIn().setBody(event.newValue());
+ } else {
+ exchange.getIn().setHeader(resultHeader, event.newValue());
+ }
+
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ getExceptionHandler().handleException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueEndpoint.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueEndpoint.java
new file mode 100644
index 0000000..e4478e0
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueEndpoint.java
@@ -0,0 +1,61 @@
+/**
+ * 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.atomix.client.value;
+
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientEndpoint;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+
+@UriEndpoint(
+ firstVersion = "2.20.0",
+ scheme = "atomix-value",
+ title = "Atomix Value",
+ syntax = "atomix-value:valueName",
+ consumerClass = AtomixValueConsumer.class,
+ label = "clustering")
+final class AtomixValueEndpoint extends AbstractAtomixClientEndpoint<AtomixValueComponent, AtomixValueConfiguration> {
+ @UriParam
+ private AtomixValueConfiguration configuration;
+
+ public AtomixValueEndpoint(String uri, AtomixValueComponent component, String resourceName) {
+ super(uri, component, resourceName);
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new AtomixValueProducer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ return new AtomixValueConsumer(this, processor, getResourceName());
+ }
+
+ @Override
+ public AtomixValueConfiguration getConfiguration() {
+ return this.configuration;
+ }
+
+ @Override
+ public void setConfiguration(AtomixValueConfiguration configuration) {
+ this.configuration = configuration;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueProducer.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueProducer.java
new file mode 100644
index 0000000..c6ca5c4
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/value/AtomixValueProducer.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.atomix.client.value;
+
+import java.time.Duration;
+
+import io.atomix.resource.ReadConsistency;
+import io.atomix.variables.DistributedValue;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.InvokeOnHeader;
+import org.apache.camel.Message;
+import org.apache.camel.component.atomix.client.AbstractAtomixClientProducer;
+import org.apache.camel.util.ObjectHelper;
+
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_ACTION;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_NAME;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_OLD_VALUE;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_READ_CONSISTENCY;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_TTL;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_VALUE;
+
+final class AtomixValueProducer extends AbstractAtomixClientProducer<AtomixValueEndpoint, DistributedValue> {
+ private final AtomixValueConfiguration configuration;
+
+ protected AtomixValueProducer(AtomixValueEndpoint endpoint) {
+ super(endpoint);
+ this.configuration = endpoint.getConfiguration();
+ }
+
+ // *********************************
+ // Handlers
+ // *********************************
+
+ @InvokeOnHeader("SET")
+ boolean onSet(Message message, AsyncCallback callback) throws Exception {
+ final DistributedValue<Object> value = getResource(message);
+ final long ttl = message.getHeader(RESOURCE_TTL, configuration::getTtl, long.class);
+ final Object val = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+
+ ObjectHelper.notNull(val, RESOURCE_VALUE);
+
+ if (ttl > 0) {
+ value.set(val, Duration.ofMillis(ttl)).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ value.set(val).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("GET")
+ boolean onGet(Message message, AsyncCallback callback) throws Exception {
+ final DistributedValue<Object> value = getResource(message);
+ final ReadConsistency consistency = message.getHeader(RESOURCE_READ_CONSISTENCY, configuration::getReadConsistency, ReadConsistency.class);
+
+ if (consistency != null) {
+ value.get(consistency).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ value.get().thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("GET_AND_SET")
+ boolean onGetAndSet(Message message, AsyncCallback callback) throws Exception {
+ final DistributedValue<Object> value = getResource(message);
+ final long ttl = message.getHeader(RESOURCE_TTL, configuration::getTtl, long.class);
+ final Object val = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+
+ ObjectHelper.notNull(val, RESOURCE_VALUE);
+
+ if (ttl > 0) {
+ value.getAndSet(val, Duration.ofMillis(ttl)).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ value.getAndSet(val).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ @InvokeOnHeader("COMPARE_AND_SET")
+ boolean onCompareAndSet(Message message, AsyncCallback callback) throws Exception {
+ final DistributedValue<Object> value = getResource(message);
+ final long ttl = message.getHeader(RESOURCE_TTL, configuration::getTtl, long.class);
+ final Object newVal = message.getHeader(RESOURCE_VALUE, message::getBody, Object.class);
+ final Object oldVal = message.getHeader(RESOURCE_OLD_VALUE, Object.class);
+
+ ObjectHelper.notNull(newVal, RESOURCE_VALUE);
+ ObjectHelper.notNull(oldVal, RESOURCE_OLD_VALUE);
+
+ if (ttl > 0) {
+ value.compareAndSet(oldVal, newVal, Duration.ofMillis(ttl)).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ } else {
+ value.compareAndSet(oldVal, newVal).thenAccept(
+ result -> processResult(message, callback, result)
+ );
+ }
+
+ return false;
+ }
+
+ // *********************************
+ // Implementation
+ // *********************************
+
+ @Override
+ protected String getProcessorKey(Message message) {
+ return message.getHeader(RESOURCE_ACTION, configuration::getDefaultAction, String.class);
+ }
+
+ @Override
+ protected String getResourceName(Message message) {
+ return message.getHeader(RESOURCE_NAME, getAtomixEndpoint()::getResourceName, String.class);
+ }
+
+ @Override
+ protected DistributedValue<Object> createResource(String resourceName) {
+ return getAtomixEndpoint()
+ .getAtomix()
+ .getValue(
+ resourceName,
+ new DistributedValue.Config(getAtomixEndpoint().getConfiguration().getResourceOptions(resourceName)),
+ new DistributedValue.Options(getAtomixEndpoint().getConfiguration().getResourceConfig(resourceName)))
+ .join();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterComponent.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterComponent.java
deleted file mode 100644
index 8db6487..0000000
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterComponent.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/**
- * 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.atomix.cluster;
-
-import java.util.Map;
-
-import org.apache.camel.CamelContext;
-import org.apache.camel.Endpoint;
-import org.apache.camel.impl.DefaultComponent;
-
-public class AtomixClusterComponent extends DefaultComponent {
- public AtomixClusterComponent() {
- }
-
- public AtomixClusterComponent(CamelContext camelContext) {
- super(camelContext);
- }
-
- @Override
- protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
- throw new UnsupportedOperationException("Not yet implemented");
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterConfiguration.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterConfiguration.java
deleted file mode 100644
index b05230f..0000000
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterConfiguration.java
+++ /dev/null
@@ -1,105 +0,0 @@
-/**
- * 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.atomix.cluster;
-
-import io.atomix.AtomixReplica;
-import io.atomix.catalyst.transport.Transport;
-import io.atomix.copycat.server.storage.StorageLevel;
-import org.apache.camel.RuntimeCamelException;
-import org.apache.camel.component.atomix.AtomixConfiguration;
-import org.apache.camel.spi.UriParam;
-import org.apache.camel.spi.UriParams;
-
-@UriParams
-public class AtomixClusterConfiguration extends AtomixConfiguration<AtomixReplica> implements Cloneable {
-
- @UriParam
- private Class<? extends Transport> clientTransport;
-
- @UriParam
- private Class<? extends Transport> serverTransport;
-
- @UriParam
- private String storagePath;
-
- @UriParam(defaultValue = "MEMORY")
- private StorageLevel storageLevel = StorageLevel.MEMORY;
-
- public AtomixClusterConfiguration() {
- }
-
- // ******************************************
- // Properties
- // ******************************************
-
-
- public Class<? extends Transport> getClientTransport() {
- return clientTransport;
- }
-
- /**
- * The client transport
- */
- public void setClientTransport(Class<? extends Transport> clientTransport) {
- this.clientTransport = clientTransport;
- }
-
- public Class<? extends Transport> getServerTransport() {
- return serverTransport;
- }
-
- /**
- * The server transport
- */
- public void setServerTransport(Class<? extends Transport> serverTransport) {
- this.serverTransport = serverTransport;
- }
-
- public String getStoragePath() {
- return storagePath;
- }
-
- /**
- * Sets the log directory.
- */
- public void setStoragePath(String storagePath) {
- this.storagePath = storagePath;
- }
-
- public StorageLevel getStorageLevel() {
- return storageLevel;
- }
-
- /**
- * Sets the log storage level.
- */
- public void setStorageLevel(StorageLevel storageLevel) {
- this.storageLevel = storageLevel;
- }
-
- // ****************************************
- // Copy
- // ****************************************
-
- public AtomixClusterConfiguration copy() {
- try {
- return (AtomixClusterConfiguration) super.clone();
- } catch (CloneNotSupportedException e) {
- throw new RuntimeCamelException(e);
- }
- }
-}
\ No newline at end of file