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:31 UTC
[6/6] camel git commit: CAMEL-10054: Create camel-atomix component
CAMEL-10054: Create camel-atomix component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/3756fba2
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/3756fba2
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/3756fba2
Branch: refs/heads/master
Commit: 3756fba22402ad81caf59f7c25859e5a95737e65
Parents: e4e7aba
Author: lburgazzoli <lb...@gmail.com>
Authored: Sat Jun 17 16:25:02 2017 +0200
Committer: lburgazzoli <lb...@gmail.com>
Committed: Wed Jun 21 18:26:35 2017 +0200
----------------------------------------------------------------------
components/camel-atomix/pom.xml | 28 +-
.../src/main/docs/atomix-cluster-component.adoc | 14 -
.../src/main/docs/atomix-map-component.adoc | 156 ++++++-
.../main/docs/atomix-messaging-component.adoc | 84 ++++
.../main/docs/atomix-multimap-component.adoc | 83 ++++
.../src/main/docs/atomix-queue-component.adoc | 81 ++++
.../src/main/docs/atomix-set-component.adoc | 82 ++++
.../src/main/docs/atomix-value-component.adoc | 84 ++++
.../component/atomix/AtomixConfiguration.java | 117 ++++-
.../component/atomix/AtomixTypeConverter.java | 32 ++
.../AbstractAsyncAtomixClientProducer.java | 140 ------
.../client/AbstractAtomixClientComponent.java | 40 ++
.../client/AbstractAtomixClientEndpoint.java | 39 +-
.../client/AbstractAtomixClientProducer.java | 134 +++++-
.../atomix/client/AtomixClientAction.java | 32 --
.../client/AtomixClientConfiguration.java | 18 +
.../atomix/client/AtomixClientConstants.java | 5 +
.../client/map/AtomixClientMapComponent.java | 67 ---
.../map/AtomixClientMapConfiguration.java | 109 -----
.../client/map/AtomixClientMapConsumer.java | 85 ----
.../client/map/AtomixClientMapEndpoint.java | 53 ---
.../client/map/AtomixClientMapProducer.java | 363 ----------------
.../component/atomix/client/map/AtomixMap.java | 38 ++
.../atomix/client/map/AtomixMapComponent.java | 70 +++
.../client/map/AtomixMapConfiguration.java | 82 ++++
.../atomix/client/map/AtomixMapConsumer.java | 102 +++++
.../atomix/client/map/AtomixMapEndpoint.java | 61 +++
.../atomix/client/map/AtomixMapProducer.java | 336 +++++++++++++++
.../client/messaging/AtomixMessaging.java | 58 +++
.../messaging/AtomixMessagingComponent.java | 70 +++
.../messaging/AtomixMessagingConfiguration.java | 94 ++++
.../messaging/AtomixMessagingConsumer.java | 121 ++++++
.../messaging/AtomixMessagingEndpoint.java | 62 +++
.../messaging/AtomixMessagingProducer.java | 117 +++++
.../atomix/client/multimap/AtomixMultiMap.java | 36 ++
.../multimap/AtomixMultiMapComponent.java | 70 +++
.../multimap/AtomixMultiMapConfiguration.java | 82 ++++
.../client/multimap/AtomixMultiMapEndpoint.java | 54 +++
.../client/multimap/AtomixMultiMapProducer.java | 277 ++++++++++++
.../atomix/client/queue/AtomixQueue.java | 35 ++
.../client/queue/AtomixQueueComponent.java | 70 +++
.../client/queue/AtomixQueueConfiguration.java | 55 +++
.../client/queue/AtomixQueueConsumer.java | 92 ++++
.../client/queue/AtomixQueueEndpoint.java | 61 +++
.../client/queue/AtomixQueueProducer.java | 204 +++++++++
.../component/atomix/client/set/AtomixSet.java | 32 ++
.../atomix/client/set/AtomixSetComponent.java | 70 +++
.../client/set/AtomixSetConfiguration.java | 68 +++
.../atomix/client/set/AtomixSetConsumer.java | 92 ++++
.../atomix/client/set/AtomixSetEndpoint.java | 61 +++
.../atomix/client/set/AtomixSetProducer.java | 174 ++++++++
.../atomix/client/value/AtomixValue.java | 30 ++
.../client/value/AtomixValueComponent.java | 70 +++
.../client/value/AtomixValueConfiguration.java | 68 +++
.../client/value/AtomixValueConsumer.java | 92 ++++
.../client/value/AtomixValueEndpoint.java | 61 +++
.../client/value/AtomixValueProducer.java | 155 +++++++
.../atomix/cluster/AtomixClusterComponent.java | 37 --
.../cluster/AtomixClusterConfiguration.java | 105 -----
.../atomix/cluster/AtomixClusterHelper.java | 83 ----
.../atomix/ha/AtomixClusterConfiguration.java | 105 +++++
.../atomix/ha/AtomixClusterHelper.java | 83 ++++
.../atomix/ha/AtomixClusterService.java | 2 -
.../component/atomix/ha/AtomixClusterView.java | 11 +-
.../services/org/apache/camel/TypeConverter | 18 +
.../org/apache/camel/component/atomix-cluster | 18 -
.../org/apache/camel/component/atomix-map | 2 +-
.../org/apache/camel/component/atomix-messaging | 18 +
.../org/apache/camel/component/atomix-multimap | 18 +
.../org/apache/camel/component/atomix-queue | 18 +
.../org/apache/camel/component/atomix-set | 18 +
.../org/apache/camel/component/atomix-value | 18 +
.../atomix/AtomixTypeConverterTest.java | 48 +++
.../client/AtomixClientSpringTestSupport.java | 36 ++
.../atomix/client/AtomixClientTestSupport.java | 35 +-
.../component/atomix/client/AtomixFactory.java | 47 ++
.../client/map/AtomixClientMapConsumerTest.java | 115 -----
.../client/map/AtomixClientMapProducerTest.java | 398 -----------------
.../client/map/AtomixMapConsumerTest.java | 136 ++++++
.../client/map/AtomixMapNodesProducerTest.java | 100 +++++
.../client/map/AtomixMapProducerTest.java | 425 +++++++++++++++++++
.../client/map/SpringAtomixMapProducerTest.java | 54 +++
.../client/messaging/AtomixMessagingTest.java | 98 +++++
.../multimap/AtomixMultiMapProducerTest.java | 352 +++++++++++++++
.../client/queue/AtomixQueueConsumerTest.java | 108 +++++
.../client/queue/AtomixQueueProducerTest.java | 235 ++++++++++
.../client/set/AtomixSetConsumerTest.java | 108 +++++
.../client/set/AtomixSetProducerTest.java | 204 +++++++++
.../client/value/AtomixValueConsumerTest.java | 98 +++++
.../client/value/AtomixValueProducerTest.java | 127 ++++++
.../client/map/SpringAtomixMapProducerTest.xml | 65 +++
...omixClientMapComponentAutoConfiguration.java | 129 ------
.../AtomixClientMapComponentConfiguration.java | 144 -------
.../AtomixMapComponentAutoConfiguration.java | 128 ++++++
.../AtomixMapComponentConfiguration.java | 158 +++++++
...omixMessagingComponentAutoConfiguration.java | 129 ++++++
.../AtomixMessagingComponentConfiguration.java | 170 ++++++++
...tomixMultiMapComponentAutoConfiguration.java | 129 ++++++
.../AtomixMultiMapComponentConfiguration.java | 158 +++++++
.../AtomixQueueComponentAutoConfiguration.java | 129 ++++++
.../AtomixQueueComponentConfiguration.java | 133 ++++++
.../AtomixSetComponentAutoConfiguration.java | 128 ++++++
.../AtomixSetComponentConfiguration.java | 145 +++++++
.../AtomixValueComponentAutoConfiguration.java | 129 ++++++
.../AtomixValueComponentConfiguration.java | 145 +++++++
.../main/resources/META-INF/spring.factories | 24 +-
106 files changed, 8454 insertions(+), 1933 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-atomix/pom.xml b/components/camel-atomix/pom.xml
index 26d808e..577280c 100644
--- a/components/camel-atomix/pom.xml
+++ b/components/camel-atomix/pom.xml
@@ -37,12 +37,21 @@
<camel.osgi.export.pkg>
org.apache.camel.component.atomix.*,
org.apache.camel.component.atomix.client.*,
- org.apache.camel.component.atomix.cluster.*
+ org.apache.camel.component.atomix.client.map.*,
+ org.apache.camel.component.atomix.client.multimap.*,
+ org.apache.camel.component.atomix.client.set.*,
+ org.apache.camel.component.atomix.client.queue.*,
+ org.apache.camel.component.atomix.client.value.*,
+ org.apache.camel.component.atomix.client.messaging.*,
org.apache.camel.component.atomix.ha.*
</camel.osgi.export.pkg>
<camel.osgi.export.service>
org.apache.camel.spi.ComponentResolver;component=atomix-map,
- org.apache.camel.spi.ComponentResolver;component=atomix-cluster,
+ org.apache.camel.spi.ComponentResolver;component=atomix-multimap,
+ org.apache.camel.spi.ComponentResolver;component=atomix-set,
+ org.apache.camel.spi.ComponentResolver;component=atomix-queue,
+ org.apache.camel.spi.ComponentResolver;component=atomix-value,
+ org.apache.camel.spi.ComponentResolver;component=atomix-messaging
</camel.osgi.export.service>
</properties>
@@ -71,6 +80,21 @@
<version>${atomix-version}</version>
</dependency>
<dependency>
+ <groupId>io.atomix</groupId>
+ <artifactId>atomix-variables</artifactId>
+ <version>${atomix-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.atomix</groupId>
+ <artifactId>atomix-concurrent</artifactId>
+ <version>${atomix-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.atomix</groupId>
+ <artifactId>atomix-resource</artifactId>
+ <version>${atomix-version}</version>
+ </dependency>
+ <dependency>
<groupId>io.atomix.catalyst</groupId>
<artifactId>catalyst-netty</artifactId>
<version>${atomix-catalyst-version}</version>
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/docs/atomix-cluster-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/docs/atomix-cluster-component.adoc b/components/camel-atomix/src/main/docs/atomix-cluster-component.adoc
deleted file mode 100644
index 5ae5df1..0000000
--- a/components/camel-atomix/src/main/docs/atomix-cluster-component.adoc
+++ /dev/null
@@ -1,14 +0,0 @@
-## Atomix Cluster Component
-
-*Available as of Camel version 2.20*
-
-The camel atomix component allows you to work with Atomix, a fault-tolerant distributed coordination framework.
-
-
-mvn \
- -Dexec.mainClass=org.apache.camel.component.atomix.ha.AtomixRoutePolicyMain \
- -Dexec.classpathScope=test \
- test-compile \
- exec:java \
- -Datomix.cluster="127.0.0.1:9001,127.0.0.1:9002,127.0.0.1:9003" \
- -Datomix.index="0"
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/docs/atomix-map-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/docs/atomix-map-component.adoc b/components/camel-atomix/src/main/docs/atomix-map-component.adoc
index d8c5262..b558857 100644
--- a/components/camel-atomix/src/main/docs/atomix-map-component.adoc
+++ b/components/camel-atomix/src/main/docs/atomix-map-component.adoc
@@ -4,6 +4,27 @@
The camel atomix component allows you to work with Atomix, a fault-tolerant distributed coordination framework.
+Maven users will need to add the following dependency to their pom.xml
+for this component:
+
+[source,java]
+----
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-atomix</artifactId>
+ <version>${camel-version}</version>
+ </dependency>
+----
+
+### URI format
+
+[source,java]
+----
+ atomix-map:mapName
+----
+
+### Options
+
// component options: START
The Atomix Map component supports 5 options which are listed below.
@@ -12,10 +33,10 @@ The Atomix Map component supports 5 options which are listed below.
[width="100%",cols="2,5,^1,2",options="header"]
|=======================================================================
| Name | Description | Default | Type
-| **configuration** (producer) | The shared component configuration | | AtomixClientMap Configuration
-| **atomix** (producer) | The shared AtomixClient instance | | AtomixClient
-| **nodes** (producer) | The nodes the AtomixClient should connect to | | List
-| **configurationUri** (producer) | The path to the AtomixClient configuration | | String
+| **configuration** (common) | The shared component configuration | | AtomixMapConfiguration
+| **atomix** (common) | The shared AtomixClient instance | | AtomixClient
+| **nodes** (common) | The nodes the AtomixClient should connect to | | List
+| **configurationUri** (common) | The path to the AtomixClient configuration | | String
| **resolveProperty Placeholders** (advanced) | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders. | true | boolean
|=======================================================================
// component options: END
@@ -23,7 +44,10 @@ The Atomix Map component supports 5 options which are listed below.
// endpoint options: START
The Atomix Map endpoint is configured using URI syntax:
+[source,java]
+----
atomix-map:mapName
+----
with the following path and query parameters:
@@ -32,14 +56,134 @@ with the following path and query parameters:
[width="100%",cols="2,5,^1,2",options="header"]
|=======================================================================
| Name | Description | Default | Type
-| **mapName** | *Required* The distributed map name | | String
+| **resourceName** | *Required* The distributed resource name | | String
|=======================================================================
-#### Query Parameters (1 parameters):
+#### Query Parameters (17 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|=======================================================================
| Name | Description | Default | Type
+| **atomix** (common) | The Atomix instance to use | | T
+| **configurationUri** (common) | The Atomix configuration uri. | | String
+| **defaultAction** (common) | The default action. | PUT | Action
+| **key** (common) | The key to use if none is set in the header or to listen for events for a specific key. | | Object
+| **nodes** (common) | The address of the nodes composing the cluster. | | String
+| **resultHeader** (common) | The header that wil carry the result. | | String
+| **transport** (common) | Sets the Atomix transport. | io.atomix.catalyst.transport.netty.NettyTransport | Transport>
+| **ttl** (common) | The resource ttl. | | long
+| **bridgeErrorHandler** (consumer) | Allows for bridging the consumer to the Camel routing Error Handler which mean any exceptions occurred while the consumer is trying to pickup incoming messages or the likes will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions that will be logged at WARN or ERROR level and ignored. | false | boolean
+| **exceptionHandler** (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this options is not in use. By default the consumer will deal with exceptions that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
+| **exchangePattern** (consumer) | Sets the exchange pattern when the consumer creates an exchange. | | ExchangePattern
+| **defaultResourceConfig** (advanced) | The cluster wide default resource configuration. | | Properties
+| **defaultResourceOptions** (advanced) | The local default resource options. | | Properties
+| **readConsistency** (advanced) | The read consistency level. | | ReadConsistency
+| **resourceConfigs** (advanced) | Cluster wide resources configuration. | | Map
+| **resourceOptions** (advanced) | Local resources configurations | | Map
| **synchronous** (advanced) | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported). | false | boolean
|=======================================================================
// endpoint options: END
+
+### Headers
+
+[width="100%",cols="10%,10%,20%a,60%",options="header",]
+|===
+| Name
+| Type
+| Values
+| Description
+
+| CamelAtomixResourceAction
+| AtomixMap.Action
+| * PUT
+ * PUT_IF_ABSENT
+ * GET
+ * CLEAR
+ * SIZE
+ * CONTAINS_KEY
+ * CONTAINS_VALUE
+ * IS_EMPTY
+ * ENTRY_SET
+ * REMOVE
+ * REPLACE
+ * VALUES
+| The action to perform
+
+| CamelAtomixResourceKey
+| Object
+| -
+| The key to operate on
+
+| CamelAtomixResourceValue
+| Object
+| -
+| The value, if missing In Body is used
+
+| CamelAtomixResourceOldValue
+| Object
+| -
+| The old value
+
+| CamelAtomixResourceTTL
+| String / long
+| -
+| The entry TTL
+
+| CamelAtomixResourceReadConsistency
+| ReadConsistency
+| * ATOMIC
+ * ATOMIC_LEASE
+ * SEQUENTIAL
+ * LOCAL
+| The read consistency level
+
+|===
+
+### Configuring the component to connect to an Atomix cluster
+
+The nodes of the Atomix cluster you want to join can be se at Endpoint or component level (recommended), below some examples:
+
+* *Endpoint:*
++
+[source,xml]
+----
+<beans xmlns="...">
+ <camelContext xmlns="http://camel.apache.org/schema/spring">
+ <from uri="direct:start"/>
+ <to uri="atomix-map:myMap?nodes=node-1.atomix.cluster:8700,node-2.atomix.cluster:8700"/>
+ </route>
+ </camelContext>
+</beans>
+----
+
+* *Component:*
++
+[source,xml]
+----
+<beans xmlns="...">
+ <bean id="atomix-map" class="org.apache.camel.component.atomix.client.map.AtomixMapComponent">
+ <property name="nodes" value="nodes=node-1.atomix.cluster:8700,node-2.atomix.cluster:8700"/>
+ </bean>
+
+ <camelContext xmlns="http://camel.apache.org/schema/spring">
+ <from uri="direct:start"/>
+ <to uri="atomix-map:myMap"/>
+ </route>
+ </camelContext>
+</beans>
+----
+
+### Usage examples:
+
+* *PUT an element with TTL of 1 second:*
++
+[source,java]
+----
+FluentProducerTemplate.on(context)
+ .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.PUT)
+ .withHeader(AtomixClientConstants.RESOURCE_KEY, key)
+ .withHeader(AtomixClientConstants.RESOURCE_TTL, "1s")
+ .withBody(val)
+ .to("direct:start")
+ .send();
+----
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/docs/atomix-messaging-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/docs/atomix-messaging-component.adoc b/components/camel-atomix/src/main/docs/atomix-messaging-component.adoc
new file mode 100644
index 0000000..dfaac7c
--- /dev/null
+++ b/components/camel-atomix/src/main/docs/atomix-messaging-component.adoc
@@ -0,0 +1,84 @@
+## Atomix Messaging Component
+
+*Available as of Camel version 2.20*
+
+The camel atomix component allows you to work with Atomix, a fault-tolerant distributed coordination framework.
+
+Maven users will need to add the following dependency to their pom.xml
+for this component:
+
+[source,java]
+----
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-atomix</artifactId>
+ <version>${camel-version}</version>
+ </dependency>
+----
+
+### URI format
+
+[source,java]
+----
+ atomix-messaging:group
+----
+
+// component options: START
+The Atomix Messaging component supports 5 options which are listed below.
+
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **configuration** (common) | The shared component configuration | | AtomixMessaging Configuration
+| **atomix** (common) | The shared AtomixClient instance | | AtomixClient
+| **nodes** (common) | The nodes the AtomixClient should connect to | | List
+| **configurationUri** (common) | The path to the AtomixClient configuration | | String
+| **resolveProperty Placeholders** (advanced) | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders. | true | boolean
+|=======================================================================
+// component options: END
+
+// endpoint options: START
+The Atomix Messaging endpoint is configured using URI syntax:
+
+[source,java]
+----
+ atomix-messaging:group
+----
+
+with the following path and query parameters:
+
+#### Path Parameters (1 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **resourceName** | *Required* The distributed resource name | | String
+|=======================================================================
+
+#### Query Parameters (18 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **atomix** (common) | The Atomix instance to use | | T
+| **broadcastType** (common) | The broadcast type. | ALL | BroadcastType
+| **channelName** (common) | The messaging channel name | | String
+| **configurationUri** (common) | The Atomix configuration uri. | | String
+| **defaultAction** (common) | The default action. | DIRECT | Action
+| **memberName** (common) | The Atomix Group member name | | String
+| **nodes** (common) | The address of the nodes composing the cluster. | | String
+| **resultHeader** (common) | The header that wil carry the result. | | String
+| **transport** (common) | Sets the Atomix transport. | io.atomix.catalyst.transport.netty.NettyTransport | Transport>
+| **bridgeErrorHandler** (consumer) | Allows for bridging the consumer to the Camel routing Error Handler which mean any exceptions occurred while the consumer is trying to pickup incoming messages or the likes will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions that will be logged at WARN or ERROR level and ignored. | false | boolean
+| **exceptionHandler** (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this options is not in use. By default the consumer will deal with exceptions that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
+| **exchangePattern** (consumer) | Sets the exchange pattern when the consumer creates an exchange. | | ExchangePattern
+| **defaultResourceConfig** (advanced) | The cluster wide default resource configuration. | | Properties
+| **defaultResourceOptions** (advanced) | The local default resource options. | | Properties
+| **readConsistency** (advanced) | The read consistency level. | | ReadConsistency
+| **resourceConfigs** (advanced) | Cluster wide resources configuration. | | Map
+| **resourceOptions** (advanced) | Local resources configurations | | Map
+| **synchronous** (advanced) | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported). | false | boolean
+|=======================================================================
+// endpoint options: END
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/docs/atomix-multimap-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/docs/atomix-multimap-component.adoc b/components/camel-atomix/src/main/docs/atomix-multimap-component.adoc
new file mode 100644
index 0000000..684029a
--- /dev/null
+++ b/components/camel-atomix/src/main/docs/atomix-multimap-component.adoc
@@ -0,0 +1,83 @@
+## Atomix MultiMap Component
+
+*Available as of Camel version 2.20*
+
+The camel atomix component allows you to work with Atomix, a fault-tolerant distributed coordination framework.
+
+Maven users will need to add the following dependency to their pom.xml
+for this component:
+
+[source,java]
+----
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-atomix</artifactId>
+ <version>${camel-version}</version>
+ </dependency>
+----
+
+### URI format
+
+[source,java]
+----
+ atomix-multimap:multiMapName
+----
+
+// component options: START
+The Atomix MultiMap component supports 5 options which are listed below.
+
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **configuration** (consumer) | The shared component configuration | | AtomixMultiMap Configuration
+| **atomix** (consumer) | The shared AtomixClient instance | | AtomixClient
+| **nodes** (consumer) | The nodes the AtomixClient should connect to | | List
+| **configurationUri** (consumer) | The path to the AtomixClient configuration | | String
+| **resolveProperty Placeholders** (advanced) | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders. | true | boolean
+|=======================================================================
+// component options: END
+
+// endpoint options: START
+The Atomix MultiMap endpoint is configured using URI syntax:
+
+[source,java]
+----
+ atomix-multimap:multiMapName
+----
+
+with the following path and query parameters:
+
+#### Path Parameters (1 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **resourceName** | *Required* The distributed resource name | | String
+|=======================================================================
+
+#### Query Parameters (17 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **atomix** (consumer) | The Atomix instance to use | | T
+| **bridgeErrorHandler** (consumer) | Allows for bridging the consumer to the Camel routing Error Handler which mean any exceptions occurred while the consumer is trying to pickup incoming messages or the likes will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions that will be logged at WARN or ERROR level and ignored. | false | boolean
+| **configurationUri** (consumer) | The Atomix configuration uri. | | String
+| **defaultAction** (consumer) | The default action. | PUT | Action
+| **key** (consumer) | The key to use if none is set in the header or to listen for events for a specific key. | | Object
+| **nodes** (consumer) | The address of the nodes composing the cluster. | | String
+| **resultHeader** (consumer) | The header that wil carry the result. | | String
+| **transport** (consumer) | Sets the Atomix transport. | io.atomix.catalyst.transport.netty.NettyTransport | Transport>
+| **ttl** (consumer) | The resource ttl. | | long
+| **exceptionHandler** (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this options is not in use. By default the consumer will deal with exceptions that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
+| **exchangePattern** (consumer) | Sets the exchange pattern when the consumer creates an exchange. | | ExchangePattern
+| **defaultResourceConfig** (advanced) | The cluster wide default resource configuration. | | Properties
+| **defaultResourceOptions** (advanced) | The local default resource options. | | Properties
+| **readConsistency** (advanced) | The read consistency level. | | ReadConsistency
+| **resourceConfigs** (advanced) | Cluster wide resources configuration. | | Map
+| **resourceOptions** (advanced) | Local resources configurations | | Map
+| **synchronous** (advanced) | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported). | false | boolean
+|=======================================================================
+// endpoint options: END
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/docs/atomix-queue-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/docs/atomix-queue-component.adoc b/components/camel-atomix/src/main/docs/atomix-queue-component.adoc
new file mode 100644
index 0000000..cff0b9b
--- /dev/null
+++ b/components/camel-atomix/src/main/docs/atomix-queue-component.adoc
@@ -0,0 +1,81 @@
+## Atomix Queue Component
+
+*Available as of Camel version 2.20*
+
+The camel atomix component allows you to work with Atomix, a fault-tolerant distributed coordination framework.
+
+Maven users will need to add the following dependency to their pom.xml
+for this component:
+
+[source,java]
+----
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-atomix</artifactId>
+ <version>${camel-version}</version>
+ </dependency>
+----
+
+### URI format
+
+[source,java]
+----
+ atomix-queue:queueName
+----
+
+// component options: START
+The Atomix Queue component supports 5 options which are listed below.
+
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **configuration** (common) | The shared component configuration | | AtomixQueue Configuration
+| **atomix** (common) | The shared AtomixClient instance | | AtomixClient
+| **nodes** (common) | The nodes the AtomixClient should connect to | | List
+| **configurationUri** (common) | The path to the AtomixClient configuration | | String
+| **resolveProperty Placeholders** (advanced) | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders. | true | boolean
+|=======================================================================
+// component options: END
+
+// endpoint options: START
+The Atomix Queue endpoint is configured using URI syntax:
+
+[source,java]
+----
+ atomix-queue:queueName
+---
+
+with the following path and query parameters:
+
+#### Path Parameters (1 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **resourceName** | *Required* The distributed resource name | | String
+|=======================================================================
+
+#### Query Parameters (15 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **atomix** (common) | The Atomix instance to use | | T
+| **configurationUri** (common) | The Atomix configuration uri. | | String
+| **defaultAction** (common) | The default action. | ADD | Action
+| **nodes** (common) | The address of the nodes composing the cluster. | | String
+| **resultHeader** (common) | The header that wil carry the result. | | String
+| **transport** (common) | Sets the Atomix transport. | io.atomix.catalyst.transport.netty.NettyTransport | Transport>
+| **bridgeErrorHandler** (consumer) | Allows for bridging the consumer to the Camel routing Error Handler which mean any exceptions occurred while the consumer is trying to pickup incoming messages or the likes will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions that will be logged at WARN or ERROR level and ignored. | false | boolean
+| **exceptionHandler** (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this options is not in use. By default the consumer will deal with exceptions that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
+| **exchangePattern** (consumer) | Sets the exchange pattern when the consumer creates an exchange. | | ExchangePattern
+| **defaultResourceConfig** (advanced) | The cluster wide default resource configuration. | | Properties
+| **defaultResourceOptions** (advanced) | The local default resource options. | | Properties
+| **readConsistency** (advanced) | The read consistency level. | | ReadConsistency
+| **resourceConfigs** (advanced) | Cluster wide resources configuration. | | Map
+| **resourceOptions** (advanced) | Local resources configurations | | Map
+| **synchronous** (advanced) | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported). | false | boolean
+|=======================================================================
+// endpoint options: END
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/docs/atomix-set-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/docs/atomix-set-component.adoc b/components/camel-atomix/src/main/docs/atomix-set-component.adoc
new file mode 100644
index 0000000..2299fe2
--- /dev/null
+++ b/components/camel-atomix/src/main/docs/atomix-set-component.adoc
@@ -0,0 +1,82 @@
+## Atomix Set Component
+
+*Available as of Camel version 2.20*
+
+The camel atomix component allows you to work with Atomix, a fault-tolerant distributed coordination framework.
+
+Maven users will need to add the following dependency to their pom.xml
+for this component:
+
+[source,java]
+----
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-atomix</artifactId>
+ <version>${camel-version}</version>
+ </dependency>
+----
+
+### URI format
+
+[source,java]
+----
+ atomix-set:setName
+----
+
+// component options: START
+The Atomix Set component supports 5 options which are listed below.
+
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **configuration** (common) | The shared component configuration | | AtomixSetConfiguration
+| **atomix** (common) | The shared AtomixClient instance | | AtomixClient
+| **nodes** (common) | The nodes the AtomixClient should connect to | | List
+| **configurationUri** (common) | The path to the AtomixClient configuration | | String
+| **resolveProperty Placeholders** (advanced) | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders. | true | boolean
+|=======================================================================
+// component options: END
+
+// endpoint options: START
+The Atomix Set endpoint is configured using URI syntax:
+
+[source,java]
+----
+ atomix-set:setName
+----
+
+with the following path and query parameters:
+
+#### Path Parameters (1 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **resourceName** | *Required* The distributed resource name | | String
+|=======================================================================
+
+#### Query Parameters (16 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **atomix** (common) | The Atomix instance to use | | T
+| **configurationUri** (common) | The Atomix configuration uri. | | String
+| **defaultAction** (common) | The default action. | ADD | Action
+| **nodes** (common) | The address of the nodes composing the cluster. | | String
+| **resultHeader** (common) | The header that wil carry the result. | | String
+| **transport** (common) | Sets the Atomix transport. | io.atomix.catalyst.transport.netty.NettyTransport | Transport>
+| **ttl** (common) | The resource ttl. | | long
+| **bridgeErrorHandler** (consumer) | Allows for bridging the consumer to the Camel routing Error Handler which mean any exceptions occurred while the consumer is trying to pickup incoming messages or the likes will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions that will be logged at WARN or ERROR level and ignored. | false | boolean
+| **exceptionHandler** (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this options is not in use. By default the consumer will deal with exceptions that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
+| **exchangePattern** (consumer) | Sets the exchange pattern when the consumer creates an exchange. | | ExchangePattern
+| **defaultResourceConfig** (advanced) | The cluster wide default resource configuration. | | Properties
+| **defaultResourceOptions** (advanced) | The local default resource options. | | Properties
+| **readConsistency** (advanced) | The read consistency level. | | ReadConsistency
+| **resourceConfigs** (advanced) | Cluster wide resources configuration. | | Map
+| **resourceOptions** (advanced) | Local resources configurations | | Map
+| **synchronous** (advanced) | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported). | false | boolean
+|=======================================================================
+// endpoint options: END
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/docs/atomix-value-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/docs/atomix-value-component.adoc b/components/camel-atomix/src/main/docs/atomix-value-component.adoc
new file mode 100644
index 0000000..8cc36d4
--- /dev/null
+++ b/components/camel-atomix/src/main/docs/atomix-value-component.adoc
@@ -0,0 +1,84 @@
+## Atomix Value Component
+
+*Available as of Camel version 2.20*
+
+The camel atomix component allows you to work with Atomix, a fault-tolerant distributed coordination framework.
+
+Maven users will need to add the following dependency to their pom.xml
+for this component:
+
+[source,java]
+----
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-atomix</artifactId>
+ <version>${camel-version}</version>
+ </dependency>
+----
+
+### URI format
+
+[source,java]
+----
+ atomix-value:valueName
+----
+
+// component options: START
+The Atomix Value component supports 5 options which are listed below.
+
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **configuration** (common) | The shared component configuration | | AtomixValue Configuration
+| **atomix** (common) | The shared AtomixClient instance | | AtomixClient
+| **nodes** (common) | The nodes the AtomixClient should connect to | | List
+| **configurationUri** (common) | The path to the AtomixClient configuration | | String
+| **resolveProperty Placeholders** (advanced) | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders. | true | boolean
+|=======================================================================
+// component options: END
+
+// endpoint options: START
+The Atomix Value endpoint is configured using URI syntax:
+
+### URI format
+
+[source,java]
+----
+ atomix-value:valueName
+----
+
+with the following path and query parameters:
+
+#### Path Parameters (1 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **resourceName** | *Required* The distributed resource name | | String
+|=======================================================================
+
+#### Query Parameters (16 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|=======================================================================
+| Name | Description | Default | Type
+| **atomix** (common) | The Atomix instance to use | | T
+| **configurationUri** (common) | The Atomix configuration uri. | | String
+| **defaultAction** (common) | The default action. | SET | Action
+| **nodes** (common) | The address of the nodes composing the cluster. | | String
+| **resultHeader** (common) | The header that wil carry the result. | | String
+| **transport** (common) | Sets the Atomix transport. | io.atomix.catalyst.transport.netty.NettyTransport | Transport>
+| **ttl** (common) | The resource ttl. | | long
+| **bridgeErrorHandler** (consumer) | Allows for bridging the consumer to the Camel routing Error Handler which mean any exceptions occurred while the consumer is trying to pickup incoming messages or the likes will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions that will be logged at WARN or ERROR level and ignored. | false | boolean
+| **exceptionHandler** (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this options is not in use. By default the consumer will deal with exceptions that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
+| **exchangePattern** (consumer) | Sets the exchange pattern when the consumer creates an exchange. | | ExchangePattern
+| **defaultResourceConfig** (advanced) | The cluster wide default resource configuration. | | Properties
+| **defaultResourceOptions** (advanced) | The local default resource options. | | Properties
+| **readConsistency** (advanced) | The read consistency level. | | ReadConsistency
+| **resourceConfigs** (advanced) | Cluster wide resources configuration. | | Map
+| **resourceOptions** (advanced) | Local resources configurations | | Map
+| **synchronous** (advanced) | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported). | false | boolean
+|=======================================================================
+// endpoint options: END
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/AtomixConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/AtomixConfiguration.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/AtomixConfiguration.java
index 1ffc06d..ee2cf0f 100644
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/AtomixConfiguration.java
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/AtomixConfiguration.java
@@ -17,7 +17,10 @@
package org.apache.camel.component.atomix;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Properties;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -32,18 +35,22 @@ import org.apache.camel.util.ObjectHelper;
public class AtomixConfiguration<T extends Atomix> implements Cloneable {
@UriParam
private T atomix;
-
@UriParam(javaType = "java.lang.String")
private List<Address> nodes = Collections.emptyList();
-
@UriParam(defaultValue = "io.atomix.catalyst.transport.netty.NettyTransport")
private Class<? extends Transport> transport = NettyTransport.class;
-
@UriParam
private String configurationUri;
-
@UriParam(label = "advanced")
private ReadConsistency readConsistency;
+ @UriParam(label = "advanced")
+ private Properties defaultResourceConfig;
+ @UriParam(label = "advanced")
+ private Properties defaultResourceOptions;
+ @UriParam(label = "advanced", prefix = "resource.config")
+ private Map<String, Properties> resourceConfigs;
+ @UriParam(label = "advanced", prefix = "resource.options")
+ private Map<String, Properties> resourceOptions;
protected AtomixConfiguration() {
}
@@ -112,4 +119,106 @@ public class AtomixConfiguration<T extends Atomix> implements Cloneable {
public void setReadConsistency(ReadConsistency readConsistency) {
this.readConsistency = readConsistency;
}
+
+ // ***********************************
+ // Properties - Resource configuration
+ // ***********************************
+
+ public Properties getDefaultResourceConfig() {
+ return defaultResourceConfig;
+ }
+
+ /**
+ * The cluster wide default resource configuration.
+ */
+ public void setDefaultResourceConfig(Properties defaultResourceConfig) {
+ this.defaultResourceConfig = defaultResourceConfig;
+ }
+
+ public Properties getDefaultResourceOptions() {
+ return defaultResourceOptions;
+ }
+
+ /**
+ * The local default resource options.
+ */
+ public void setDefaultResourceOptions(Properties defaultResourceOptions) {
+ this.defaultResourceOptions = defaultResourceOptions;
+ }
+
+ public Map<String, Properties> getResourceConfigs() {
+ return resourceConfigs;
+ }
+
+ /**
+ * Cluster wide resources configuration.
+ */
+ public void setResourceConfigs(Map<String, Properties> resourceConfigs) {
+ this.resourceConfigs = resourceConfigs;
+ }
+
+ public void addResourceConfig(String name, Properties config) {
+ if (this.resourceConfigs == null) {
+ this.resourceConfigs = new HashMap<>();
+ }
+
+ this.resourceConfigs.put(name, config);
+ }
+
+ public Properties getResourceConfig(String name) {
+ Properties properties = null;
+
+ if (this.resourceConfigs != null) {
+ Properties props = this.resourceConfigs.getOrDefault(name, this.defaultResourceConfig);
+ if (props != null) {
+ properties = new Properties(props);
+ }
+ } else if (this.defaultResourceConfig != null) {
+ properties = new Properties(this.defaultResourceConfig);
+ }
+
+ if (properties == null) {
+ properties = new Properties();
+ }
+
+ return properties;
+ }
+
+ public Map<String, Properties> getResourceOptions() {
+ return resourceOptions;
+ }
+
+ /**
+ * Local resources configurations
+ */
+ public void setResourceOptions(Map<String, Properties> resourceOptions) {
+ this.resourceOptions = resourceOptions;
+ }
+
+ public void addResourceOption(String name, Properties config) {
+ if (this.resourceOptions == null) {
+ this.resourceOptions = new HashMap<>();
+ }
+
+ this.resourceOptions.put(name, config);
+ }
+
+ public Properties getResourceOptions(String name) {
+ Properties properties = null;
+
+ if (this.resourceOptions != null) {
+ Properties props = this.resourceOptions.getOrDefault(name, this.defaultResourceOptions);
+ if (props != null) {
+ properties = new Properties(props);
+ }
+ } else if (this.defaultResourceOptions != null) {
+ properties = new Properties(this.defaultResourceOptions);
+ }
+
+ if (properties == null) {
+ properties = new Properties();
+ }
+
+ return properties;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/AtomixTypeConverter.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/AtomixTypeConverter.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/AtomixTypeConverter.java
new file mode 100644
index 0000000..3f2c9c1
--- /dev/null
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/AtomixTypeConverter.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;
+
+import io.atomix.catalyst.transport.Address;
+import org.apache.camel.Converter;
+
+@Converter
+public final class AtomixTypeConverter {
+
+ private AtomixTypeConverter() {
+ }
+
+ @Converter
+ public static Address toAddress(String address) {
+ return new Address(address);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAsyncAtomixClientProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAsyncAtomixClientProducer.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAsyncAtomixClientProducer.java
deleted file mode 100644
index 931e9e4..0000000
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAsyncAtomixClientProducer.java
+++ /dev/null
@@ -1,140 +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.client;
-
-import java.lang.annotation.ElementType;
-import java.lang.annotation.Repeatable;
-import java.lang.annotation.Retention;
-import java.lang.annotation.RetentionPolicy;
-import java.lang.annotation.Target;
-import java.lang.reflect.Method;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.camel.AsyncCallback;
-import org.apache.camel.AsyncProcessor;
-import org.apache.camel.Exchange;
-import org.apache.camel.Message;
-import org.apache.camel.RuntimeCamelException;
-import org.apache.camel.component.atomix.AtomixAsyncMessageProcessor;
-import org.apache.camel.impl.DefaultProducer;
-import org.apache.camel.util.AsyncProcessorHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public abstract class AbstractAsyncAtomixClientProducer<E extends AbstractAtomixClientEndpoint> extends DefaultProducer implements AsyncProcessor {
- private static final Logger LOGGER = LoggerFactory.getLogger(AbstractAsyncAtomixClientProducer.class);
- private final Map<AtomixClientAction, AtomixAsyncMessageProcessor> processors;
-
- protected AbstractAsyncAtomixClientProducer(E endpoint) {
- super(endpoint);
-
- this.processors = new HashMap<>();
- }
-
- @Override
- protected void doStart() throws Exception {
- for (final Method method : getClass().getDeclaredMethods()) {
- AsyncInvokeOnHeaders annotations = method.getAnnotation(AsyncInvokeOnHeaders.class);
- if (annotations != null) {
- for (AsyncInvokeOnHeader annotation : annotations.value()) {
- bind(annotation, method);
- }
- } else {
- AsyncInvokeOnHeader annotation = method.getAnnotation(AsyncInvokeOnHeader.class);
- if (annotation != null) {
- bind(annotation, method);
- }
- }
- }
-
- super.doStart();
- }
-
- @Override
- public void process(Exchange exchange) throws Exception {
- AsyncProcessorHelper.process(this, exchange);
- }
-
- @Override
- public boolean process(Exchange exchange, AsyncCallback callback) {
- final Message message = exchange.getIn();
- final AtomixClientAction action = getAction(message);
-
- AtomixAsyncMessageProcessor processor = this.processors.get(action);
- if (processor != null) {
- try {
- return processor.process(message, callback);
- } catch (Exception e) {
- throw new RuntimeCamelException(e);
- }
- } else {
- throw new RuntimeCamelException("No handler for action " + action);
- }
- }
-
- @SuppressWarnings("unchecked")
- protected E getAtomixEndpoint() {
- return (E)super.getEndpoint();
- }
-
- protected abstract AtomixClientAction getAction(Message message);
-
- // ************************************
- // Binding helpers
- // ************************************
-
- private void bind(AsyncInvokeOnHeader annotation, final Method method) {
- if (method.getParameterCount() == 2) {
- method.setAccessible(true);
-
- if (!Message.class.isAssignableFrom(method.getParameterTypes()[0])) {
- throw new IllegalArgumentException("First argument should be of type Message");
- }
- if (!AsyncCallback.class.isAssignableFrom(method.getParameterTypes()[1])) {
- throw new IllegalArgumentException("Second argument should be of type AsyncCallback");
- }
-
- LOGGER.debug("bind key={}, class={}, method={}",
- annotation.value(), this.getClass(), method.getName());
-
- this.processors.put(annotation.value(), (m, c) -> (boolean)method.invoke(this, m, c));
- } else {
- throw new IllegalArgumentException(
- "Illegal number of parameters for method: " + method.getName() + ", required: 2, found: " + method.getParameterCount()
- );
- }
- }
-
- // ************************************
- // Annotations
- // ************************************
-
- @Repeatable(AsyncInvokeOnHeaders.class)
- @Retention(RetentionPolicy.RUNTIME)
- @Target(ElementType.METHOD)
- public @interface AsyncInvokeOnHeader {
- AtomixClientAction value();
- }
-
- @Retention(RetentionPolicy.RUNTIME)
- @Target(ElementType.METHOD)
- public @interface AsyncInvokeOnHeaders {
- AsyncInvokeOnHeader[] value();
- }
-}
-
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientComponent.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientComponent.java
index 9048b3b..642ce65 100644
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientComponent.java
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientComponent.java
@@ -17,11 +17,15 @@
package org.apache.camel.component.atomix.client;
import java.util.List;
+import java.util.Map;
+import java.util.Properties;
import io.atomix.AtomixClient;
import io.atomix.catalyst.transport.Address;
import org.apache.camel.CamelContext;
import org.apache.camel.impl.DefaultComponent;
+import org.apache.camel.util.EndpointHelper;
+import org.apache.camel.util.IntrospectionSupport;
public abstract class AbstractAtomixClientComponent<C extends AtomixClientConfiguration> extends DefaultComponent {
protected AbstractAtomixClientComponent() {
@@ -76,6 +80,42 @@ public abstract class AbstractAtomixClientComponent<C extends AtomixClientConfig
// *****************************************
// Properties
// *****************************************
+
+ protected C setConfigurationProperties(C configuration, Map<String, Object> parameters) throws Exception {
+ // Resolve config for named maps
+ Map<String, Object> configs = IntrospectionSupport.extractProperties(parameters, "resource.config.");
+ for (Map.Entry<String, Object> entry : configs.entrySet()) {
+ String ref = (String)entry.getValue();
+ if (!EndpointHelper.isReferenceParameter(ref)) {
+ throw new IllegalArgumentException("The option resource.config." + ref + " should be a reference");
+ }
+
+ configuration.addResourceConfig(
+ entry.getKey(),
+ EndpointHelper.resolveReferenceParameter(getCamelContext(), ref, Properties.class));
+ }
+
+ // Resolve options for named maps
+ Map<String, Object> options = IntrospectionSupport.extractProperties(parameters, "resource.options.");
+ for (Map.Entry<String, Object> entry : options.entrySet()) {
+ String ref = (String)entry.getValue();
+ if (!EndpointHelper.isReferenceParameter(ref)) {
+ throw new IllegalArgumentException("The option resource.options." + ref + " should be a reference");
+ }
+
+ configuration.addResourceOption(
+ entry.getKey(),
+ EndpointHelper.resolveReferenceParameter(getCamelContext(), ref, Properties.class));
+ }
+
+ setProperties(configuration, parameters);
+
+ return configuration;
+ }
+
+ // *****************************************
+ // Properties
+ // *****************************************
protected abstract C getComponentConfiguration();
}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientEndpoint.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientEndpoint.java
index 5a83873..93abc1a 100644
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientEndpoint.java
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientEndpoint.java
@@ -17,20 +17,27 @@
package org.apache.camel.component.atomix.client;
import io.atomix.AtomixClient;
+import org.apache.camel.CamelContext;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriPath;
+import org.apache.camel.util.ObjectHelper;
public abstract class AbstractAtomixClientEndpoint<T extends AbstractAtomixClientComponent, C extends AtomixClientConfiguration> extends DefaultEndpoint {
- private final C configuration;
+ @UriPath(description = "The distributed resource name")
+ @Metadata(required = "true")
+ private final String resourceName;
+
private AtomixClient atomix;
- protected AbstractAtomixClientEndpoint(String uri, T component, C configuration) {
+ protected AbstractAtomixClientEndpoint(String uri, T component, String resourceName) {
super(uri, component);
- this.configuration = configuration;
+ this.resourceName = resourceName;
}
@Override
@@ -51,7 +58,13 @@ public abstract class AbstractAtomixClientEndpoint<T extends AbstractAtomixClien
@Override
protected void doStart() throws Exception {
if (atomix == null) {
- atomix = AtomixClientHelper.createClient(getCamelContext(), configuration);
+ final C configuration = getConfiguration();
+ final CamelContext context = getCamelContext();
+
+ ObjectHelper.notNull(configuration, "Configuration");
+ ObjectHelper.notNull(context, "CamelContext");
+
+ atomix = AtomixClientHelper.createClient(context, configuration);
atomix.connect(configuration.getNodes()).join();
}
@@ -69,18 +82,26 @@ public abstract class AbstractAtomixClientEndpoint<T extends AbstractAtomixClien
// **********************************
// Helpers for implementations
- // *********************************
+ // **********************************
@SuppressWarnings("unchecked")
public T getAtomixComponent() {
return (T)super.getComponent();
}
- public C getAtomixConfiguration() {
- return this.configuration;
- }
-
public AtomixClient getAtomix() {
return atomix;
}
+
+ public String getResourceName() {
+ return resourceName;
+ }
+
+ // **********************************
+ // Abstract
+ // **********************************
+
+ public abstract C getConfiguration();
+
+ public abstract void setConfiguration(C configuration);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientProducer.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientProducer.java
index 16a7866..21e785f 100644
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientProducer.java
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AbstractAtomixClientProducer.java
@@ -16,15 +16,141 @@
*/
package org.apache.camel.component.atomix.client;
-import org.apache.camel.impl.HeaderSelectorProducer;
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
-public abstract class AbstractAtomixClientProducer<E extends AbstractAtomixClientEndpoint> extends HeaderSelectorProducer {
- protected AbstractAtomixClientProducer(E endpoint, String header) {
- super(endpoint, header);
+import io.atomix.resource.Resource;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.Exchange;
+import org.apache.camel.InvokeOnHeader;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.atomix.AtomixAsyncMessageProcessor;
+import org.apache.camel.impl.DefaultProducer;
+import org.apache.camel.util.AsyncProcessorHelper;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT;
+import static org.apache.camel.component.atomix.client.AtomixClientConstants.RESOURCE_NAME;
+
+public abstract class AbstractAtomixClientProducer<E extends AbstractAtomixClientEndpoint, R extends Resource> extends DefaultProducer implements AsyncProcessor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(AbstractAtomixClientProducer.class);
+ private final Map<String, AtomixAsyncMessageProcessor> processors;
+ private ConcurrentMap<String, R> resources;
+
+ protected AbstractAtomixClientProducer(E endpoint) {
+ super(endpoint);
+
+ this.processors = new HashMap<>();
+ this.resources = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ for (final Method method : getClass().getDeclaredMethods()) {
+ InvokeOnHeader[] annotations = method.getAnnotationsByType(InvokeOnHeader.class);
+ if (annotations != null && annotations.length > 0) {
+ for (InvokeOnHeader annotation : annotations) {
+ bind(annotation, method);
+ }
+ }
+ }
+
+ super.doStart();
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ AsyncProcessorHelper.process(this, exchange);
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ final Message message = exchange.getIn();
+ final String key = getProcessorKey(message);
+
+ AtomixAsyncMessageProcessor processor = this.processors.get(key);
+ if (processor != null) {
+ try {
+ return processor.process(message, callback);
+ } catch (Exception e) {
+ throw new RuntimeCamelException(e);
+ }
+ } else {
+ throw new RuntimeCamelException("No handler for action " + key);
+ }
}
+ // **********************************
+ //
+ // **********************************
+
@SuppressWarnings("unchecked")
protected E getAtomixEndpoint() {
return (E)super.getEndpoint();
}
+
+ protected void processResult(Message message, AsyncCallback callback, Object result) {
+ if (result != null && !(result instanceof Void)) {
+ message.setHeader(RESOURCE_ACTION_HAS_RESULT, true);
+
+ String resultHeader = getAtomixEndpoint().getConfiguration().getResultHeader();
+ if (resultHeader != null) {
+ message.setHeader(resultHeader, result);
+ } else {
+ message.setBody(result);
+ }
+ } else {
+ message.setHeader(RESOURCE_ACTION_HAS_RESULT, false);
+ }
+
+ callback.done(false);
+ }
+
+ protected R getResource(Message message) {
+ String resourceName = getResourceName(message);
+
+ ObjectHelper.notNull(resourceName, RESOURCE_NAME);
+
+ return resources.computeIfAbsent(resourceName, name -> createResource(name));
+ }
+
+ protected abstract String getProcessorKey(Message message);
+
+ protected abstract String getResourceName(Message message);
+
+ protected abstract R createResource(String name);
+
+ // ************************************
+ // Binding helpers
+ // ************************************
+
+ private void bind(InvokeOnHeader annotation, final Method method) {
+ if (method.getParameterCount() == 2) {
+ method.setAccessible(true);
+
+ if (!Message.class.isAssignableFrom(method.getParameterTypes()[0])) {
+ throw new IllegalArgumentException("First argument should be of type Message");
+ }
+ if (!AsyncCallback.class.isAssignableFrom(method.getParameterTypes()[1])) {
+ throw new IllegalArgumentException("Second argument should be of type AsyncCallback");
+ }
+
+ LOGGER.debug("bind key={}, class={}, method={}",
+ annotation.value(), this.getClass(), method.getName());
+
+ this.processors.put(annotation.value(), (m, c) -> (boolean)method.invoke(this, m, c));
+ } else {
+ throw new IllegalArgumentException(
+ "Illegal number of parameters for method: " + method.getName() + ", required: 2, found: " + method.getParameterCount()
+ );
+ }
+ }
}
+
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientAction.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientAction.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientAction.java
deleted file mode 100644
index 5bf641f..0000000
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientAction.java
+++ /dev/null
@@ -1,32 +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.client;
-
-public enum AtomixClientAction {
- PUT,
- PUT_IF_ABSENT,
- GET,
- CLEAR,
- SIZE,
- CONTAINS_KEY,
- CONTAINS_VALUE,
- IS_EMPTY,
- ENTRY_SET,
- REMOVE,
- REPLACE,
- VALUES
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientConfiguration.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientConfiguration.java
index 0b8a0d8..4791bde 100644
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientConfiguration.java
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientConfiguration.java
@@ -18,6 +18,24 @@ package org.apache.camel.component.atomix.client;
import io.atomix.AtomixClient;
import org.apache.camel.component.atomix.AtomixConfiguration;
+import org.apache.camel.spi.UriParam;
public class AtomixClientConfiguration extends AtomixConfiguration<AtomixClient> implements Cloneable {
+ @UriParam
+ private String resultHeader;
+
+ // ****************************************
+ // Properties
+ // ****************************************
+
+ public String getResultHeader() {
+ return resultHeader;
+ }
+
+ /**
+ * The header that wil carry the result.
+ */
+ public void setResultHeader(String resultHeader) {
+ this.resultHeader = resultHeader;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientConstants.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientConstants.java
index 2fccd8e..5cd8213 100644
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientConstants.java
+++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/AtomixClientConstants.java
@@ -27,6 +27,11 @@ public final class AtomixClientConstants {
public static final String RESOURCE_TTL = "CamelAtomixResourceTTL";
public static final String RESOURCE_READ_CONSISTENCY = "CamelAtomixResourceReadConsistency";
public static final String EVENT_TYPE = "CamelAtomixEventType";
+ public static final String MESSAGE_ID = "CamelAtomixEventType";
+ public static final String MEMBER_NAME = "CamelAtomixMemberName";
+ public static final String CHANNEL_NAME = "CamelAtomixChannelName";
+
+ public static final String BROADCAST_TYPE = "CamelAtomixBroadcastType";
private AtomixClientConstants() {
}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapComponent.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapComponent.java
deleted file mode 100644
index 7c22fbf..0000000
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapComponent.java
+++ /dev/null
@@ -1,67 +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.client.map;
-
-import java.util.Map;
-
-import org.apache.camel.CamelContext;
-import org.apache.camel.Endpoint;
-import org.apache.camel.component.atomix.client.AbstractAtomixClientComponent;
-
-public class AtomixClientMapComponent extends AbstractAtomixClientComponent<AtomixClientMapConfiguration> {
- private AtomixClientMapConfiguration configuration = new AtomixClientMapConfiguration();
-
- public AtomixClientMapComponent() {
- super();
- }
-
- public AtomixClientMapComponent(CamelContext camelContext) {
- super(camelContext);
- }
-
- // **********************************************
- // Endpoints
- // **********************************************
-
- @Override
- protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
- AtomixClientMapConfiguration configuration = this.configuration.copy();
- setProperties(configuration, parameters);
-
- return new AtomixClientMapEndpoint(uri, this, configuration, remaining);
- }
-
- // **********************************************
- // Properties
- // **********************************************
-
- public AtomixClientMapConfiguration getConfiguration() {
- return this.configuration;
- }
-
- /**
- * The shared component configuration
- */
- public void setConfiguration(AtomixClientMapConfiguration configuration) {
- this.configuration = configuration;
- }
-
- @Override
- protected AtomixClientMapConfiguration 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/map/AtomixClientMapConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapConfiguration.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapConfiguration.java
deleted file mode 100644
index ca909ac..0000000
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapConfiguration.java
+++ /dev/null
@@ -1,109 +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.client.map;
-
-import org.apache.camel.RuntimeCamelException;
-import org.apache.camel.component.atomix.client.AtomixClientAction;
-import org.apache.camel.component.atomix.client.AtomixClientConfiguration;
-import org.apache.camel.spi.UriParam;
-import org.apache.camel.spi.UriParams;
-
-@UriParams
-public class AtomixClientMapConfiguration extends AtomixClientConfiguration {
- @UriParam(defaultValue = "PUT")
- private AtomixClientAction defaultAction = AtomixClientAction.PUT;
- @UriParam
- private Long ttl;
- @UriParam
- private String resultHeader;
-// @UriParam(label = "advanced")
-// private DistributedMap.Config config = new DistributedMap.Config();
-// @UriParam(label = "advanced")
-// private DistributedMap.Options options = new DistributedMap.Options();
-
- // ****************************************
- // Properties
- // ****************************************
-
- public AtomixClientAction getDefaultAction() {
- return defaultAction;
- }
-
- /**
- * The default action.
- */
- public void setDefaultAction(AtomixClientAction defaultAction) {
- this.defaultAction = defaultAction;
- }
-
- public Long getTtl() {
- return ttl;
- }
-
- /**
- * The resource ttl.
- */
- public void setTtl(Long ttl) {
- this.ttl = ttl;
- }
-
- public String getResultHeader() {
- return resultHeader;
- }
-
- /**
- * The header that wil carry the result.
- */
- public void setResultHeader(String resultHeader) {
- this.resultHeader = resultHeader;
- }
-
-
-// public DistributedMap.Config getConfig() {
-// return config;
-// }
-//
-// /**
-// * The cluster wide map config
-// */
-// public void setConfig(DistributedMap.Config config) {
-// this.config = config;
-// }
-//
-// public DistributedMap.Options getOptions() {
-// return options;
-// }
-//
-// /**
-// * The local map options
-// */
-// public void setOptions(DistributedMap.Options options) {
-// this.options = options;
-// }
-
- // ****************************************
- // Copy
- // ****************************************
-
- public AtomixClientMapConfiguration copy() {
- try {
- return (AtomixClientMapConfiguration) 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/map/AtomixClientMapConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapConsumer.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapConsumer.java
deleted file mode 100644
index 909d92d..0000000
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapConsumer.java
+++ /dev/null
@@ -1,85 +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.client.map;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import io.atomix.catalyst.concurrent.Listener;
-import io.atomix.collections.DistributedMap;
-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;
-
-public class AtomixClientMapConsumer extends AbstractAtomixClientConsumer<AtomixClientMapEndpoint> {
- private final List<Listener<DistributedMap.EntryEvent<Object, Object>>> listeners;
- private DistributedMap<Object, Object> map;
-
- public AtomixClientMapConsumer(AtomixClientMapEndpoint endpoint, Processor processor) {
- super(endpoint, processor);
- this.listeners = new ArrayList<>();
- }
-
- @Override
- protected void doStart() throws Exception {
- super.doStart();
-
- this.map = getAtomixEndpoint()
- .getAtomix()
- .getMap(
- getAtomixEndpoint().getMapName())
- //getAtomixEndpoint().getAtomixConfiguration().getConfig(),
- //getAtomixEndpoint().getAtomixConfiguration().getOptions())
- .join();
-
- this.listeners.add(this.map.onAdd(this::onEvent).join());
- this.listeners.add(this.map.onRemove(this::onEvent).join());
- this.listeners.add(this.map.onUpdate(this::onEvent).join());
- }
-
- @Override
- protected void doStop() throws Exception {
- // close listeners
- listeners.forEach(Listener::close);
-
- // close the map
- if (this.map == null) {
- this.map.close().join();
- this.map = null;
- }
-
- super.doStart();
- }
-
- // ********************************************
- // Event handler
- // ********************************************
-
- private void onEvent(DistributedMap.EntryEvent<Object, Object> event) {
- Exchange exchange = getEndpoint().createExchange();
- exchange.getIn().setHeader(AtomixClientConstants.EVENT_TYPE, event.type());
- exchange.getIn().setHeader(AtomixClientConstants.RESOURCE_KEY, event.entry().getKey());
- exchange.getIn().setBody(event.entry().getValue());
-
- 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/map/AtomixClientMapEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapEndpoint.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapEndpoint.java
deleted file mode 100644
index fc48660..0000000
--- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/client/map/AtomixClientMapEndpoint.java
+++ /dev/null
@@ -1,53 +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.client.map;
-
-import org.apache.camel.Producer;
-import org.apache.camel.component.atomix.client.AbstractAtomixClientEndpoint;
-import org.apache.camel.spi.Metadata;
-import org.apache.camel.spi.UriEndpoint;
-import org.apache.camel.spi.UriPath;
-
-@UriEndpoint(firstVersion = "2.20.0", scheme = "atomix-map", title = "Atomix Map", syntax = "atomix-map:mapName", producerOnly = true, label = "clustering")
-class AtomixClientMapEndpoint extends AbstractAtomixClientEndpoint<AtomixClientMapComponent, AtomixClientMapConfiguration> {
-
- @UriPath(description = "The distributed map name")
- @Metadata(required = "true")
- private final String mapName;
-
- public AtomixClientMapEndpoint(String uri, AtomixClientMapComponent component, AtomixClientMapConfiguration configuration, String mapName) {
- super(uri, component, configuration);
-
- this.mapName = mapName;
- }
-
- @Override
- public Producer createProducer() throws Exception {
- return new AtomixClientMapProducer(this, mapName);
- }
-
- /*
- @Override
- public Consumer createConsumer(Processor processor) throws Exception {
- return new AtomixClientMapConsumer(this, processor);
- }
- */
-
- public String getMapName() {
- return mapName;
- }
-}