This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 17ba42af2df0 CAMEL-26236: camel-ignite - the existing query results
carry the key header, a PUT of a Map with a key stores it as one value, RUN
keeps the body, and no operation or no body fails with a clear message (#29078)
17ba42af2df0 is described below
commit 17ba42af2df0101be81d0884be96d2343b49c163
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Nov 3 12:11:28 2026 +0100
CAMEL-26236: camel-ignite - the existing query results carry the key
header, a PUT of a Map with a key stores it as one value, RUN keeps the body,
and no operation or no body fails with a clear message (#29078)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/components/ignite-cache.json | 6 +-
.../camel/catalog/components/ignite-compute.json | 4 +-
.../camel/catalog/components/ignite-idgen.json | 2 +-
.../camel/catalog/components/ignite-queue.json | 2 +-
.../camel/catalog/docs/ignite-cache-component.adoc | 6 +-
.../catalog/docs/ignite-compute-component.adoc | 3 +
.../camel/catalog/docs/ignite-idgen-component.adoc | 4 ++
.../camel/catalog/docs/ignite-queue-component.adoc | 3 +
.../camel/catalog/docs/ignite-set-component.adoc | 3 +
.../camel/component/ignite/cache/ignite-cache.json | 6 +-
.../component/ignite/compute/ignite-compute.json | 4 +-
.../camel/component/ignite/idgen/ignite-idgen.json | 2 +-
.../camel/component/ignite/queue/ignite-queue.json | 2 +-
.../src/main/docs/ignite-cache-component.adoc | 6 +-
.../src/main/docs/ignite-compute-component.adoc | 3 +
.../src/main/docs/ignite-idgen-component.adoc | 4 ++
.../src/main/docs/ignite-queue-component.adoc | 3 +
.../src/main/docs/ignite-set-component.adoc | 3 +
.../cache/IgniteCacheContinuousQueryConsumer.java | 9 +--
.../ignite/cache/IgniteCacheEndpoint.java | 17 +++--
.../ignite/cache/IgniteCacheProducer.java | 31 ++++++---
.../ignite/compute/IgniteComputeEndpoint.java | 15 +++--
.../ignite/compute/IgniteComputeProducer.java | 75 ++++++++++++++++------
.../ignite/idgen/IgniteIdGenEndpoint.java | 6 +-
.../ignite/idgen/IgniteIdGenProducer.java | 31 +++++++--
.../ignite/queue/IgniteQueueEndpoint.java | 8 +--
.../ignite/queue/IgniteQueueProducer.java | 33 ++++++++--
.../component/ignite/set/IgniteSetProducer.java | 33 ++++++++--
.../ignite/IgniteCacheContinuousQueryTest.java | 21 ++++++
.../camel/component/ignite/IgniteCacheTest.java | 68 ++++++++++++++++++++
.../camel/component/ignite/IgniteComputeTest.java | 63 ++++++++++++++++--
.../camel/component/ignite/IgniteIdGenTest.java | 44 +++++++++++++
.../camel/component/ignite/IgniteQueueTest.java | 29 +++++++++
.../camel/component/ignite/IgniteSetTest.java | 28 ++++++++
.../ignite/TestIgniteComputeResources.java | 26 ++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 25 ++++++++
36 files changed, 535 insertions(+), 93 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-cache.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-cache.json
index 348cb48c6c01..c229e182cc3e 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-cache.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-cache.json
@@ -41,9 +41,9 @@
"CamelIgniteCacheOldValue": { "index": 6, "kind": "header", "displayName":
"", "group": "common", "label": "", "required": false, "javaType": "Object",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "(producer) The old cache value to be replaced when
invoking the REPLACE operation. (consumer) This header carries the old cache
value when passed in the incoming cache event.", "constantName":
"org.apache.camel.component.ignite.IgniteConst [...]
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"oneExchangePerUpdate", "value": "false", "carriers": [ "list" ] }, { "option":
"fireExistingQueryResults", "value": "true", "carriers": [ "unknown" ] } ],
"description": "The new value of the cache entry that changed, with its key in
the CamelIgniteCacheKey header, the old value in CamelIgniteCacheOldValue and
the event type in CamelIgniteCacheEventType. With oneExchangePerUpdate=false a
List of the CacheEntryEven [...]
- "producerIn": { "carriers": [ "any" ], "description": "PUT and REPLACE
store the body as the value of the key in the CamelIgniteCacheKey header, and a
PUT of a Map puts all its entries; no body fails. GET and REMOVE take the key
from that header, or else from the body, where a Set is several keys. QUERY
needs an Ignite Query in the body or the CamelIgniteCacheQuery header. SIZE,
CLEAR and REBALANCE do not use the body. The operation is the operation option
or the CamelIgniteCacheOper [...]
- "producerOut": { "carriers": [ "unknown" ], "dependsOn": [
"propagateIncomingBodyIfNoReturnValue" ], "byOption": [ { "option":
"operation", "value": "PUT", "carriers": [ "unchanged", "null" ] }, { "option":
"operation", "value": "REPLACE", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "REMOVE", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "CLEAR", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "REBALANCE", "carriers": [ "unchan [...]
+ "consumerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"oneExchangePerUpdate", "value": "false", "carriers": [ "list" ] } ],
"description": "The new value of the cache entry that changed, with its key in
the CamelIgniteCacheKey header, the old value in CamelIgniteCacheOldValue and
the event type in CamelIgniteCacheEventType. With oneExchangePerUpdate=false a
List of the CacheEntryEvents of Ignite per batch; with
fireExistingQueryResults=true the entries the initial que [...]
+ "producerIn": { "carriers": [ "any" ], "description": "PUT and REPLACE
store the body as the value of the key in the CamelIgniteCacheKey header, a Map
as well. A PUT of a Map without that header puts all its entries, unless
treatCollectionsAsCacheObjects=true; no body fails. GET and REMOVE take the key
from that header, or else from the body, where a Set is several keys. QUERY
needs an Ignite Query in the body or the CamelIgniteCacheQuery header. SIZE,
CLEAR and REBALANCE do not use [...]
+ "producerOut": { "carriers": [ "unknown" ], "dependsOn": [
"propagateIncomingBodyIfNoReturnValue" ], "byOption": [ { "option":
"operation", "value": "PUT", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "REPLACE", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "REMOVE", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "CLEAR", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "REBALANCE", "carriers": [ "unchanged" ] } [...]
},
"properties": {
"cacheName": { "index": 0, "kind": "path", "displayName": "Cache Name",
"group": "common", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The cache name." },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-compute.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-compute.json
index 6e2bedb40a51..d56fb4e453bf 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-compute.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-compute.json
@@ -38,8 +38,8 @@
"CamelIgniteComputeAffinityKey": { "index": 4, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Object", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "Affinity key for the AFFINITY_CALL and
AFFINITY_RUN operations.", "constantName":
"org.apache.camel.component.ignite.IgniteConstants#IGNITE_COMPUTE_AFFINITY_KEY"
}
},
"bodyTypes": {
- "producerIn": { "carriers": [ "pojo", "list" ], "dependsOn": [ "taskName"
], "description": "The job to run on the cluster, by executionType: an
IgniteCallable for CALL and AFFINITY_CALL, an IgniteRunnable for RUN and
AFFINITY_RUN, a Collection of them for CALL and RUN, an IgniteCallable,
IgniteRunnable or IgniteClosure for BROADCAST, an IgniteClosure for APPLY, and
a ComputeTask or its Class for EXECUTE. Its parameters are in the
CamelIgniteComputeParameters header. With taskName, E [...]
- "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"executionType", "value": "RUN", "carriers": [ "null" ] }, { "option":
"executionType", "value": "AFFINITY_RUN", "carriers": [ "null" ] }, { "option":
"executionType", "value": "BROADCAST", "carriers": [ "list", "null" ] } ],
"description": "The result of the job. A Collection of callables, a broadcast
and an APPLY of a Collection of parameters give a Collection of the results,
unless an IgniteReducer in the CamelI [...]
+ "producerIn": { "carriers": [ "pojo", "list" ], "dependsOn": [ "taskName"
], "description": "The job to run on the cluster, by executionType: an
IgniteCallable for CALL and AFFINITY_CALL, an IgniteRunnable for RUN and
AFFINITY_RUN, a Collection of them for CALL and RUN, an IgniteCallable,
IgniteRunnable or IgniteClosure for BROADCAST, an IgniteClosure for APPLY, and
a ComputeTask or its Class for EXECUTE. Its parameters are in the
CamelIgniteComputeParameters header. With taskName, E [...]
+ "producerOut": { "carriers": [ "unknown" ], "dependsOn": [
"propagateIncomingBodyIfNoReturnValue" ], "byOption": [ { "option":
"executionType", "value": "RUN", "carriers": [ "unchanged" ] }, { "option":
"executionType", "value": "AFFINITY_RUN", "carriers": [ "unchanged" ] }, {
"option": "executionType", "value": "BROADCAST", "carriers": [ "list", "null" ]
} ], "description": "The result of the job. A Collection of callables, a
broadcast and an APPLY of a Collection of parameters give [...]
},
"properties": {
"endpointId": { "index": 0, "kind": "path", "displayName": "Endpoint Id",
"group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The endpoint ID (not
used)." },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-idgen.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-idgen.json
index 432245af793d..683e251b4acf 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-idgen.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-idgen.json
@@ -34,7 +34,7 @@
"CamelIgniteIdGenOperation": { "index": 0, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "org.apache.camel.component.ignite.idgen.IgniteIdGenOperation",
"enum": [ "ADD_AND_GET", "GET", "GET_AND_ADD", "GET_AND_INCREMENT",
"INCREMENT_AND_GET" ], "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "Allows you to dynamically change the ID
Generator operation.", "constantName": "org [...]
},
"bodyTypes": {
- "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "ADD_AND_GET", "carriers": [ "value", "text" ] }, {
"option": "operation", "value": "GET_AND_ADD", "carriers": [ "value", "text" ]
} ], "description": "ADD_AND_GET and GET_AND_ADD add the body to the sequence:
a number, or text of one; no body fails. The other operations do not use the
body, but text that is not a number still fails. The operation is the operation
option or the CamelIgniteIdGenOper [...]
+ "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "ADD_AND_GET", "carriers": [ "value", "text" ] }, {
"option": "operation", "value": "GET_AND_ADD", "carriers": [ "value", "text" ]
} ], "description": "ADD_AND_GET and GET_AND_ADD add the body to the sequence:
a number, or text of one; no body fails. The other operations do not use the
body. The operation is the operation option or the CamelIgniteIdGenOperation
header; with neither the producer fails" },
"producerOut": { "carriers": [ "value" ], "javaType": "java.lang.Long",
"description": "The value of the sequence, as a Long" }
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-queue.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-queue.json
index cad10bfde3ea..1756e068812f 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-queue.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ignite-queue.json
@@ -37,7 +37,7 @@
"CamelIgniteQueueTimeoutMillis": { "index": 3, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Long", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "Dynamically sets the timeout in
milliseconds to use when invoking the OFFER or POLL operations.",
"constantName":
"org.apache.camel.component.ignite.IgniteConstants#IGNITE_QUEUE_TIMEOUT_MILLIS"
}
},
"bodyTypes": {
- "producerIn": { "carriers": [ "any" ], "description": "ADD, OFFER, PUT,
CONTAINS, REMOVE and RETAIN_ALL use the body as the element, and a Collection
as several elements unless treatCollectionsAsCacheObjects=true; no body fails
ADD, CONTAINS, REMOVE and RETAIN_ALL. DRAIN drains into a Collection body, or a
new List. The other operations do not use the body. The operation is the
operation option or the CamelIgniteQueueOperation header; with neither the
producer fails" },
+ "producerIn": { "carriers": [ "any" ], "description": "ADD, OFFER, PUT,
CONTAINS, REMOVE and RETAIN_ALL use the body as the element, and a Collection
as several elements unless treatCollectionsAsCacheObjects=true; no body fails
them. DRAIN drains into a Collection body, or a new List. The other operations
do not use the body. The operation is the operation option or the
CamelIgniteQueueOperation header; with neither the producer fails" },
"producerOut": { "carriers": [ "unknown" ], "dependsOn": [
"propagateIncomingBodyIfNoReturnValue" ], "byOption": [ { "option":
"operation", "value": "ADD", "carriers": [ "value" ], "javaType":
"java.lang.Boolean" }, { "option": "operation", "value": "OFFER", "carriers": [
"value" ], "javaType": "java.lang.Boolean" }, { "option": "operation", "value":
"CONTAINS", "carriers": [ "value" ], "javaType": "java.lang.Boolean" }, {
"option": "operation", "value": "REMOVE", "carriers": [ "valu [...]
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-cache-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-cache-component.adoc
index f25f4943ac2a..31217b1ba174 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-cache-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-cache-component.adoc
@@ -18,9 +18,13 @@ The Ignite Cache endpoint is one of camel-ignite endpoints
that allow you to int
This offers both a Producer (to invoke cache operations on an Ignite cache)
and a Consumer (to consume changes from a continuous query).
The cache value is the body of the message, and the cache key is in the
`CamelIgniteCacheKey` message header.
-GET and REMOVE take the key from the body when that header is not set, and a
PUT of a `Map` body puts all its entries.
+GET and REMOVE take the key from the body when that header is not set.
+A PUT of a `Map` body without that header puts all its entries; with the
header, or with `treatCollectionsAsCacheObjects=true`, the `Map` is stored as
one value.
Even if you configure a fixed operation in the endpoint URI, you can vary it
per-exchange by setting the `CamelIgniteCacheOperation` message header.
+Without either, the producer fails.
+
+With `fireExistingQueryResults=true`, the consumer first sends the entries
that the initial query finds, each with the value as the body and the key in
the `CamelIgniteCacheKey` header, like the events that follow.
// component options: START
include::partial$component-configure-options.adoc[]
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-compute-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-compute-component.adoc
index cadc21011e48..343e6993f2c5 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-compute-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-compute-component.adoc
@@ -41,6 +41,9 @@ Each operation expects the indicated types:
| AFFINITY_RUN | IgniteRunnable.
|=======================================================================
+Without a body the producer fails, except EXECUTE with the `taskName` option,
which then passes the `CamelIgniteComputeParameters` header to the task.
+RUN and AFFINITY_RUN have no result: they keep the body, or leave no body when
`propagateIncomingBodyIfNoReturnValue` is `false`.
+
== Usage
The component needs an Ignite node: xref:ignite-summary.adoc[Ignite] describes
the ways to give it one.
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-idgen-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-idgen-component.adoc
index 8bc0e893e125..5d4d6a052fe2 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-idgen-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-idgen-component.adoc
@@ -16,6 +16,10 @@
The Ignite ID Generator endpoint is one of camel-ignite endpoints that allow
you to interact with https://apacheignite.readme.io/docs/id-generator[Ignite
Atomic Sequences and ID Generators].
+`ADD_AND_GET` and `GET_AND_ADD` add the number in the body to the sequence,
and fail without a body.
+The other operations do not use the body.
+The operation is the `operation` option, or the `CamelIgniteIdGenOperation`
header; without either, the producer fails.
+
// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-queue-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-queue-component.adoc
index 154b9027bc4f..2bdcd2352891 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-queue-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-queue-component.adoc
@@ -16,6 +16,9 @@
The Ignite Queue endpoint is one of camel-ignite endpoints that allow you to
interact with https://apacheignite.readme.io/docs/queue-and-set[Ignite Queue
data structures].
+`ADD`, `OFFER`, `PUT`, `CONTAINS`, `REMOVE` and `RETAIN_ALL` use the body as
the element, and fail without a body.
+The operation is the `operation` option, or the `CamelIgniteQueueOperation`
header; without either, the producer fails.
+
// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-set-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-set-component.adoc
index d8a11ce1af09..e618e93f6b40 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-set-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ignite-set-component.adoc
@@ -16,6 +16,9 @@
The Ignite Sets endpoint is one of camel-ignite endpoints that allows you to
interact with https://apacheignite.readme.io/docs/queue-and-set[Ignite Set data
structures].
+`ADD`, `CONTAINS`, `REMOVE` and `RETAIN_ALL` use the body as the element, and
fail without a body.
+The operation is the `operation` option, or the `CamelIgniteSetsOperation`
header; without either, the producer fails.
+
// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
diff --git
a/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/cache/ignite-cache.json
b/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/cache/ignite-cache.json
index 348cb48c6c01..c229e182cc3e 100644
---
a/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/cache/ignite-cache.json
+++
b/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/cache/ignite-cache.json
@@ -41,9 +41,9 @@
"CamelIgniteCacheOldValue": { "index": 6, "kind": "header", "displayName":
"", "group": "common", "label": "", "required": false, "javaType": "Object",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "(producer) The old cache value to be replaced when
invoking the REPLACE operation. (consumer) This header carries the old cache
value when passed in the incoming cache event.", "constantName":
"org.apache.camel.component.ignite.IgniteConst [...]
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"oneExchangePerUpdate", "value": "false", "carriers": [ "list" ] }, { "option":
"fireExistingQueryResults", "value": "true", "carriers": [ "unknown" ] } ],
"description": "The new value of the cache entry that changed, with its key in
the CamelIgniteCacheKey header, the old value in CamelIgniteCacheOldValue and
the event type in CamelIgniteCacheEventType. With oneExchangePerUpdate=false a
List of the CacheEntryEven [...]
- "producerIn": { "carriers": [ "any" ], "description": "PUT and REPLACE
store the body as the value of the key in the CamelIgniteCacheKey header, and a
PUT of a Map puts all its entries; no body fails. GET and REMOVE take the key
from that header, or else from the body, where a Set is several keys. QUERY
needs an Ignite Query in the body or the CamelIgniteCacheQuery header. SIZE,
CLEAR and REBALANCE do not use the body. The operation is the operation option
or the CamelIgniteCacheOper [...]
- "producerOut": { "carriers": [ "unknown" ], "dependsOn": [
"propagateIncomingBodyIfNoReturnValue" ], "byOption": [ { "option":
"operation", "value": "PUT", "carriers": [ "unchanged", "null" ] }, { "option":
"operation", "value": "REPLACE", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "REMOVE", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "CLEAR", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "REBALANCE", "carriers": [ "unchan [...]
+ "consumerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"oneExchangePerUpdate", "value": "false", "carriers": [ "list" ] } ],
"description": "The new value of the cache entry that changed, with its key in
the CamelIgniteCacheKey header, the old value in CamelIgniteCacheOldValue and
the event type in CamelIgniteCacheEventType. With oneExchangePerUpdate=false a
List of the CacheEntryEvents of Ignite per batch; with
fireExistingQueryResults=true the entries the initial que [...]
+ "producerIn": { "carriers": [ "any" ], "description": "PUT and REPLACE
store the body as the value of the key in the CamelIgniteCacheKey header, a Map
as well. A PUT of a Map without that header puts all its entries, unless
treatCollectionsAsCacheObjects=true; no body fails. GET and REMOVE take the key
from that header, or else from the body, where a Set is several keys. QUERY
needs an Ignite Query in the body or the CamelIgniteCacheQuery header. SIZE,
CLEAR and REBALANCE do not use [...]
+ "producerOut": { "carriers": [ "unknown" ], "dependsOn": [
"propagateIncomingBodyIfNoReturnValue" ], "byOption": [ { "option":
"operation", "value": "PUT", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "REPLACE", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "REMOVE", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "CLEAR", "carriers": [ "unchanged" ] }, { "option":
"operation", "value": "REBALANCE", "carriers": [ "unchanged" ] } [...]
},
"properties": {
"cacheName": { "index": 0, "kind": "path", "displayName": "Cache Name",
"group": "common", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The cache name." },
diff --git
a/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/compute/ignite-compute.json
b/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/compute/ignite-compute.json
index 6e2bedb40a51..d56fb4e453bf 100644
---
a/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/compute/ignite-compute.json
+++
b/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/compute/ignite-compute.json
@@ -38,8 +38,8 @@
"CamelIgniteComputeAffinityKey": { "index": 4, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Object", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "Affinity key for the AFFINITY_CALL and
AFFINITY_RUN operations.", "constantName":
"org.apache.camel.component.ignite.IgniteConstants#IGNITE_COMPUTE_AFFINITY_KEY"
}
},
"bodyTypes": {
- "producerIn": { "carriers": [ "pojo", "list" ], "dependsOn": [ "taskName"
], "description": "The job to run on the cluster, by executionType: an
IgniteCallable for CALL and AFFINITY_CALL, an IgniteRunnable for RUN and
AFFINITY_RUN, a Collection of them for CALL and RUN, an IgniteCallable,
IgniteRunnable or IgniteClosure for BROADCAST, an IgniteClosure for APPLY, and
a ComputeTask or its Class for EXECUTE. Its parameters are in the
CamelIgniteComputeParameters header. With taskName, E [...]
- "producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"executionType", "value": "RUN", "carriers": [ "null" ] }, { "option":
"executionType", "value": "AFFINITY_RUN", "carriers": [ "null" ] }, { "option":
"executionType", "value": "BROADCAST", "carriers": [ "list", "null" ] } ],
"description": "The result of the job. A Collection of callables, a broadcast
and an APPLY of a Collection of parameters give a Collection of the results,
unless an IgniteReducer in the CamelI [...]
+ "producerIn": { "carriers": [ "pojo", "list" ], "dependsOn": [ "taskName"
], "description": "The job to run on the cluster, by executionType: an
IgniteCallable for CALL and AFFINITY_CALL, an IgniteRunnable for RUN and
AFFINITY_RUN, a Collection of them for CALL and RUN, an IgniteCallable,
IgniteRunnable or IgniteClosure for BROADCAST, an IgniteClosure for APPLY, and
a ComputeTask or its Class for EXECUTE. Its parameters are in the
CamelIgniteComputeParameters header. With taskName, E [...]
+ "producerOut": { "carriers": [ "unknown" ], "dependsOn": [
"propagateIncomingBodyIfNoReturnValue" ], "byOption": [ { "option":
"executionType", "value": "RUN", "carriers": [ "unchanged" ] }, { "option":
"executionType", "value": "AFFINITY_RUN", "carriers": [ "unchanged" ] }, {
"option": "executionType", "value": "BROADCAST", "carriers": [ "list", "null" ]
} ], "description": "The result of the job. A Collection of callables, a
broadcast and an APPLY of a Collection of parameters give [...]
},
"properties": {
"endpointId": { "index": 0, "kind": "path", "displayName": "Endpoint Id",
"group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The endpoint ID (not
used)." },
diff --git
a/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/idgen/ignite-idgen.json
b/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/idgen/ignite-idgen.json
index 432245af793d..683e251b4acf 100644
---
a/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/idgen/ignite-idgen.json
+++
b/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/idgen/ignite-idgen.json
@@ -34,7 +34,7 @@
"CamelIgniteIdGenOperation": { "index": 0, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "org.apache.camel.component.ignite.idgen.IgniteIdGenOperation",
"enum": [ "ADD_AND_GET", "GET", "GET_AND_ADD", "GET_AND_INCREMENT",
"INCREMENT_AND_GET" ], "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "Allows you to dynamically change the ID
Generator operation.", "constantName": "org [...]
},
"bodyTypes": {
- "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "ADD_AND_GET", "carriers": [ "value", "text" ] }, {
"option": "operation", "value": "GET_AND_ADD", "carriers": [ "value", "text" ]
} ], "description": "ADD_AND_GET and GET_AND_ADD add the body to the sequence:
a number, or text of one; no body fails. The other operations do not use the
body, but text that is not a number still fails. The operation is the operation
option or the CamelIgniteIdGenOper [...]
+ "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "ADD_AND_GET", "carriers": [ "value", "text" ] }, {
"option": "operation", "value": "GET_AND_ADD", "carriers": [ "value", "text" ]
} ], "description": "ADD_AND_GET and GET_AND_ADD add the body to the sequence:
a number, or text of one; no body fails. The other operations do not use the
body. The operation is the operation option or the CamelIgniteIdGenOperation
header; with neither the producer fails" },
"producerOut": { "carriers": [ "value" ], "javaType": "java.lang.Long",
"description": "The value of the sequence, as a Long" }
},
"properties": {
diff --git
a/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/queue/ignite-queue.json
b/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/queue/ignite-queue.json
index cad10bfde3ea..1756e068812f 100644
---
a/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/queue/ignite-queue.json
+++
b/components/camel-ignite/src/generated/resources/META-INF/org/apache/camel/component/ignite/queue/ignite-queue.json
@@ -37,7 +37,7 @@
"CamelIgniteQueueTimeoutMillis": { "index": 3, "kind": "header",
"displayName": "", "group": "producer", "label": "producer", "required": false,
"javaType": "Long", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "Dynamically sets the timeout in
milliseconds to use when invoking the OFFER or POLL operations.",
"constantName":
"org.apache.camel.component.ignite.IgniteConstants#IGNITE_QUEUE_TIMEOUT_MILLIS"
}
},
"bodyTypes": {
- "producerIn": { "carriers": [ "any" ], "description": "ADD, OFFER, PUT,
CONTAINS, REMOVE and RETAIN_ALL use the body as the element, and a Collection
as several elements unless treatCollectionsAsCacheObjects=true; no body fails
ADD, CONTAINS, REMOVE and RETAIN_ALL. DRAIN drains into a Collection body, or a
new List. The other operations do not use the body. The operation is the
operation option or the CamelIgniteQueueOperation header; with neither the
producer fails" },
+ "producerIn": { "carriers": [ "any" ], "description": "ADD, OFFER, PUT,
CONTAINS, REMOVE and RETAIN_ALL use the body as the element, and a Collection
as several elements unless treatCollectionsAsCacheObjects=true; no body fails
them. DRAIN drains into a Collection body, or a new List. The other operations
do not use the body. The operation is the operation option or the
CamelIgniteQueueOperation header; with neither the producer fails" },
"producerOut": { "carriers": [ "unknown" ], "dependsOn": [
"propagateIncomingBodyIfNoReturnValue" ], "byOption": [ { "option":
"operation", "value": "ADD", "carriers": [ "value" ], "javaType":
"java.lang.Boolean" }, { "option": "operation", "value": "OFFER", "carriers": [
"value" ], "javaType": "java.lang.Boolean" }, { "option": "operation", "value":
"CONTAINS", "carriers": [ "value" ], "javaType": "java.lang.Boolean" }, {
"option": "operation", "value": "REMOVE", "carriers": [ "valu [...]
},
"properties": {
diff --git a/components/camel-ignite/src/main/docs/ignite-cache-component.adoc
b/components/camel-ignite/src/main/docs/ignite-cache-component.adoc
index f25f4943ac2a..31217b1ba174 100644
--- a/components/camel-ignite/src/main/docs/ignite-cache-component.adoc
+++ b/components/camel-ignite/src/main/docs/ignite-cache-component.adoc
@@ -18,9 +18,13 @@ The Ignite Cache endpoint is one of camel-ignite endpoints
that allow you to int
This offers both a Producer (to invoke cache operations on an Ignite cache)
and a Consumer (to consume changes from a continuous query).
The cache value is the body of the message, and the cache key is in the
`CamelIgniteCacheKey` message header.
-GET and REMOVE take the key from the body when that header is not set, and a
PUT of a `Map` body puts all its entries.
+GET and REMOVE take the key from the body when that header is not set.
+A PUT of a `Map` body without that header puts all its entries; with the
header, or with `treatCollectionsAsCacheObjects=true`, the `Map` is stored as
one value.
Even if you configure a fixed operation in the endpoint URI, you can vary it
per-exchange by setting the `CamelIgniteCacheOperation` message header.
+Without either, the producer fails.
+
+With `fireExistingQueryResults=true`, the consumer first sends the entries
that the initial query finds, each with the value as the body and the key in
the `CamelIgniteCacheKey` header, like the events that follow.
// component options: START
include::partial$component-configure-options.adoc[]
diff --git
a/components/camel-ignite/src/main/docs/ignite-compute-component.adoc
b/components/camel-ignite/src/main/docs/ignite-compute-component.adoc
index cadc21011e48..343e6993f2c5 100644
--- a/components/camel-ignite/src/main/docs/ignite-compute-component.adoc
+++ b/components/camel-ignite/src/main/docs/ignite-compute-component.adoc
@@ -41,6 +41,9 @@ Each operation expects the indicated types:
| AFFINITY_RUN | IgniteRunnable.
|=======================================================================
+Without a body the producer fails, except EXECUTE with the `taskName` option,
which then passes the `CamelIgniteComputeParameters` header to the task.
+RUN and AFFINITY_RUN have no result: they keep the body, or leave no body when
`propagateIncomingBodyIfNoReturnValue` is `false`.
+
== Usage
The component needs an Ignite node: xref:ignite-summary.adoc[Ignite] describes
the ways to give it one.
diff --git a/components/camel-ignite/src/main/docs/ignite-idgen-component.adoc
b/components/camel-ignite/src/main/docs/ignite-idgen-component.adoc
index 8bc0e893e125..5d4d6a052fe2 100644
--- a/components/camel-ignite/src/main/docs/ignite-idgen-component.adoc
+++ b/components/camel-ignite/src/main/docs/ignite-idgen-component.adoc
@@ -16,6 +16,10 @@
The Ignite ID Generator endpoint is one of camel-ignite endpoints that allow
you to interact with https://apacheignite.readme.io/docs/id-generator[Ignite
Atomic Sequences and ID Generators].
+`ADD_AND_GET` and `GET_AND_ADD` add the number in the body to the sequence,
and fail without a body.
+The other operations do not use the body.
+The operation is the `operation` option, or the `CamelIgniteIdGenOperation`
header; without either, the producer fails.
+
// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
diff --git a/components/camel-ignite/src/main/docs/ignite-queue-component.adoc
b/components/camel-ignite/src/main/docs/ignite-queue-component.adoc
index 154b9027bc4f..2bdcd2352891 100644
--- a/components/camel-ignite/src/main/docs/ignite-queue-component.adoc
+++ b/components/camel-ignite/src/main/docs/ignite-queue-component.adoc
@@ -16,6 +16,9 @@
The Ignite Queue endpoint is one of camel-ignite endpoints that allow you to
interact with https://apacheignite.readme.io/docs/queue-and-set[Ignite Queue
data structures].
+`ADD`, `OFFER`, `PUT`, `CONTAINS`, `REMOVE` and `RETAIN_ALL` use the body as
the element, and fail without a body.
+The operation is the `operation` option, or the `CamelIgniteQueueOperation`
header; without either, the producer fails.
+
// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
diff --git a/components/camel-ignite/src/main/docs/ignite-set-component.adoc
b/components/camel-ignite/src/main/docs/ignite-set-component.adoc
index d8a11ce1af09..e618e93f6b40 100644
--- a/components/camel-ignite/src/main/docs/ignite-set-component.adoc
+++ b/components/camel-ignite/src/main/docs/ignite-set-component.adoc
@@ -16,6 +16,9 @@
The Ignite Sets endpoint is one of camel-ignite endpoints that allows you to
interact with https://apacheignite.readme.io/docs/queue-and-set[Ignite Set data
structures].
+`ADD`, `CONTAINS`, `REMOVE` and `RETAIN_ALL` use the body as the element, and
fail without a body.
+The operation is the `operation` option, or the `CamelIgniteSetsOperation`
header; without either, the producer fails.
+
// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
diff --git
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java
index 04bc639e2d56..41f978f5e6a6 100644
---
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java
+++
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheContinuousQueryConsumer.java
@@ -21,7 +21,6 @@ import javax.cache.event.CacheEntryEvent;
import javax.cache.event.CacheEntryListenerException;
import javax.cache.event.CacheEntryUpdatedListener;
-import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Message;
@@ -75,14 +74,10 @@ public class IgniteCacheContinuousQueryConsumer extends
DefaultConsumer {
LOG.info("Processing existing cache results for cache name = {}.",
endpoint.getCacheName());
for (Entry<Object, Object> entry : cursor) {
+ // the value as the body and the key as a header, like the events
of the continuous query
Exchange exchange = createExchange(entry.getValue());
exchange.getIn().setHeader(IgniteConstants.IGNITE_CACHE_KEY,
entry.getKey());
- getAsyncProcessor().process(createExchange(entry), new
AsyncCallback() {
- @Override
- public void done(boolean doneSync) {
- // do nothing
- }
- });
+ getAsyncProcessor().process(exchange, EmptyAsyncCallback.get());
}
}
diff --git
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheEndpoint.java
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheEndpoint.java
index 2bff9c98b559..09484ccca4c5 100644
---
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheEndpoint.java
+++
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheEndpoint.java
@@ -50,19 +50,19 @@ import static
org.apache.camel.component.ignite.IgniteConstants.SCHEME_CACHE;
@BodyTypes(consumerOut = @BodyType(value = Carrier.UNKNOWN,
byOption = {
@BodyType.When(option =
"oneExchangePerUpdate", value = "false",
- carriers =
Carrier.LIST),
- @BodyType.When(option =
"fireExistingQueryResults", value = "true",
- carriers =
Carrier.UNKNOWN) },
+ carriers =
Carrier.LIST) },
description = "The new value of the cache
entry that changed, with its key in the"
+ " CamelIgniteCacheKey
header, the old value in CamelIgniteCacheOldValue and"
+ " the event type in
CamelIgniteCacheEventType. With oneExchangePerUpdate=false"
+ " a List of the
CacheEntryEvents of Ignite per batch; with"
+ "
fireExistingQueryResults=true the entries the initial query finds come"
- + " first, each as a JCache
Cache.Entry"),
+ + " first, each with its
value as the body and its key in the"
+ + " CamelIgniteCacheKey
header"),
producerIn = @BodyType(value = Carrier.ANY,
description = "PUT and REPLACE store the
body as the value of the key in the"
- + " CamelIgniteCacheKey
header, and a PUT of a Map puts all its entries; no"
- + " body fails. GET and REMOVE
take the key from that header, or else from the"
+ + " CamelIgniteCacheKey
header, a Map as well. A PUT of a Map without that"
+ + " header puts all its
entries, unless treatCollectionsAsCacheObjects=true;"
+ + " no body fails. GET and
REMOVE take the key from that header, or else from the"
+ " body, where a Set is
several keys. QUERY needs an Ignite Query in the body"
+ " or the
CamelIgniteCacheQuery header. SIZE, CLEAR and REBALANCE do not use"
+ " the body. The operation is
the operation option or the"
@@ -71,7 +71,7 @@ import static
org.apache.camel.component.ignite.IgniteConstants.SCHEME_CACHE;
dependsOn =
"propagateIncomingBodyIfNoReturnValue",
byOption = {
@BodyType.When(option =
"operation", value = "PUT",
- carriers = {
Carrier.UNCHANGED, Carrier.NULL }),
+ carriers =
Carrier.UNCHANGED),
@BodyType.When(option =
"operation", value = "REPLACE",
carriers =
Carrier.UNCHANGED),
@BodyType.When(option =
"operation", value = "REMOVE",
@@ -87,8 +87,7 @@ import static
org.apache.camel.component.ignite.IgniteConstants.SCHEME_CACHE;
description = "GET gives the cached value,
or null when the key is not cached, and a Map"
+ " of the keys and their
values for a Set of keys. QUERY gives an Iterator of"
+ " the entries found, closed
when the exchange completes. SIZE gives the"
- + " number of entries. The
others leave the body, but a PUT of a Map leaves no"
- + " body"))
+ + " number of entries. The
others leave the body"))
public class IgniteCacheEndpoint extends AbstractIgniteEndpoint {
@UriPath
diff --git
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheProducer.java
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheProducer.java
index afdab7542a08..abfaf30d726f 100644
---
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheProducer.java
+++
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/cache/IgniteCacheProducer.java
@@ -51,11 +51,12 @@ public class IgniteCacheProducer extends
DefaultAsyncProducer {
@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
+ IgniteCacheOperation operation = cacheOperationFor(exchange);
Message in = exchange.getIn();
Message out = exchange.getOut();
MessageHelper.copyHeaders(exchange.getIn(), out, true);
- switch (cacheOperationFor(exchange)) {
+ switch (operation) {
case GET:
doGet(in, out);
@@ -110,19 +111,23 @@ public class IgniteCacheProducer extends
DefaultAsyncProducer {
@SuppressWarnings("unchecked")
private void doPut(Message in, Message out) {
- Map<Object, Object> map = in.getBody(Map.class);
+ Object cacheKey = in.getHeader(IgniteConstants.IGNITE_CACHE_KEY);
- if (map != null) {
- cache.putAll(map);
- return;
+ // a Map is put as its entries only without a key, and when
collections are not cache objects:
+ // otherwise it is stored as one value, like any other body
+ if (cacheKey == null && !endpoint.isTreatCollectionsAsCacheObjects()) {
+ Map<Object, Object> map = in.getBody(Map.class);
+ if (map != null) {
+ cache.putAll(map);
+ IgniteHelper.maybePropagateIncomingBody(endpoint, in, out);
+ return;
+ }
}
- Object cacheKey = in.getHeader(IgniteConstants.IGNITE_CACHE_KEY);
-
if (cacheKey == null) {
throw new RuntimeCamelException(
"Cache PUT operation requires the cache key in the
CamelIgniteCacheKey header, "
- + "or a payload of type Map.");
+ + "or a payload of type Map when
treatCollectionsAsCacheObjects is false.");
}
cache.put(cacheKey, in.getBody());
@@ -227,8 +232,14 @@ public class IgniteCacheProducer extends
DefaultAsyncProducer {
}
private IgniteCacheOperation cacheOperationFor(Exchange exchange) {
- return
exchange.getIn().getHeader(IgniteConstants.IGNITE_CACHE_OPERATION,
endpoint.getOperation(),
- IgniteCacheOperation.class);
+ IgniteCacheOperation operation =
exchange.getIn().getHeader(IgniteConstants.IGNITE_CACHE_OPERATION,
+ endpoint.getOperation(), IgniteCacheOperation.class);
+ if (operation == null) {
+ throw new IllegalArgumentException(
+ "No operation specified. Set the operation via the
operation option of the endpoint or the header "
+ +
IgniteConstants.IGNITE_CACHE_OPERATION);
+ }
+ return operation;
}
}
diff --git
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java
index 136adea0c895..528d4b759602 100644
---
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java
+++
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeEndpoint.java
@@ -52,18 +52,23 @@ import static
org.apache.camel.component.ignite.IgniteConstants.SCHEME_COMPUTE;
+ " IgniteClosure for
BROADCAST, an IgniteClosure for APPLY, and a ComputeTask or"
+ " its Class for EXECUTE. Its
parameters are in the CamelIgniteComputeParameters"
+ " header. With taskName,
EXECUTE runs that task with the body as its"
- + " argument. Text or no body
fails"),
+ + " argument, or with that
header when there is no body. Otherwise text or no"
+ + " body fails"),
producerOut = @BodyType(value = Carrier.UNKNOWN,
+ dependsOn =
"propagateIncomingBodyIfNoReturnValue",
byOption = {
- @BodyType.When(option =
"executionType", value = "RUN", carriers = Carrier.NULL),
+ @BodyType.When(option =
"executionType", value = "RUN",
+ carriers =
Carrier.UNCHANGED),
@BodyType.When(option =
"executionType", value = "AFFINITY_RUN",
- carriers =
Carrier.NULL),
+ carriers =
Carrier.UNCHANGED),
@BodyType.When(option =
"executionType", value = "BROADCAST",
carriers = {
Carrier.LIST, Carrier.NULL }) },
description = "The result of the job. A
Collection of callables, a broadcast and an APPLY"
+ " of a Collection of
parameters give a Collection of the results, unless an"
- + " IgniteReducer in the
CamelIgniteComputeReducer header reduces them. RUN,"
- + " AFFINITY_RUN and a
broadcast IgniteRunnable leave no body"))
+ + " IgniteReducer in the
CamelIgniteComputeReducer header reduces them. RUN and"
+ + " AFFINITY_RUN have no
result and keep the body, or leave no body with"
+ + "
propagateIncomingBodyIfNoReturnValue=false. A broadcast IgniteRunnable"
+ + " leaves no body"))
public class IgniteComputeEndpoint extends AbstractIgniteEndpoint {
@UriPath
diff --git
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeProducer.java
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeProducer.java
index 7cd196b9d23c..db7f73ebfde5 100644
---
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeProducer.java
+++
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/compute/IgniteComputeProducer.java
@@ -21,6 +21,7 @@ import java.util.Collection;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
import org.apache.camel.NoTypeConversionAvailableException;
import org.apache.camel.RuntimeCamelException;
@@ -54,7 +55,8 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
IgniteCompute compute = endpoint.createIgniteCompute().withAsync();
try {
- switch (executionTypeFor(exchange)) {
+ IgniteComputeExecutionType executionType =
executionTypeFor(exchange);
+ switch (executionType) {
case CALL:
doCall(exchange, compute);
@@ -91,7 +93,11 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
return false;
}
- compute.future().listen(IgniteInCamelClosure.create(exchange,
callback));
+ // RUN and AFFINITY_RUN have no result, so they keep the incoming
body unless told otherwise
+ boolean propagateIncomingBody = (executionType ==
IgniteComputeExecutionType.RUN
+ || executionType ==
IgniteComputeExecutionType.AFFINITY_RUN)
+ && endpoint.isPropagateIncomingBodyIfNoReturnValue();
+ compute.future().listen(IgniteInCamelClosure.create(exchange,
callback, propagateIncomingBody));
return false;
} catch (Exception e) {
@@ -102,8 +108,9 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
}
@SuppressWarnings({ "unchecked", "rawtypes" })
- private void doCall(final Exchange exchange, IgniteCompute compute) throws
NoTypeConversionAvailableException {
- Object job = exchange.getIn().getBody();
+ private void doCall(final Exchange exchange, IgniteCompute compute)
+ throws NoTypeConversionAvailableException, InvalidPayloadException
{
+ Object job = mandatoryJob(exchange, IgniteComputeExecutionType.CALL,
"an IgniteCallable, or a Collection of them");
IgniteReducer<Object, Object> reducer
=
exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_REDUCER,
IgniteReducer.class);
@@ -131,8 +138,9 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
}
@SuppressWarnings("unchecked")
- private void doBroadcast(final Exchange exchange, IgniteCompute compute) {
- Object job = exchange.getIn().getBody();
+ private void doBroadcast(final Exchange exchange, IgniteCompute compute)
throws InvalidPayloadException {
+ Object job = mandatoryJob(exchange,
IgniteComputeExecutionType.BROADCAST,
+ "an IgniteCallable, an IgniteRunnable or an IgniteClosure");
if (IgniteCallable.class.isAssignableFrom(job.getClass())) {
compute.broadcast((IgniteCallable<?>) job);
@@ -150,20 +158,25 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
}
@SuppressWarnings("unchecked")
- private void doExecute(final Exchange exchange, IgniteCompute compute) {
+ private void doExecute(final Exchange exchange, IgniteCompute compute)
throws InvalidPayloadException {
Object job = exchange.getIn().getBody();
Object params =
exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_PARAMS);
if (job instanceof Class &&
ComputeTask.class.isAssignableFrom((Class<?>) job)) {
Class<? extends ComputeTask<Object, Object>> task = (Class<?
extends ComputeTask<Object, Object>>) job;
compute.execute(task, params);
- } else if (ComputeTask.class.isAssignableFrom(job.getClass())) {
+ } else if (job instanceof ComputeTask) {
compute.execute((ComputeTask<Object, Object>) job, params);
} else if (endpoint.getTaskName() != null) {
- if (exchange.getIn().getBody() != null) {
- params = exchange.getIn().getBody();
+ // the body is the argument of the task; without a body the
CamelIgniteComputeParameters header is
+ if (job != null) {
+ params = job;
}
compute.execute(endpoint.getTaskName(), params);
+ } else if (job == null) {
+ throw new InvalidPayloadException(
+ exchange, Object.class, exchange.getIn(),
+ "the EXECUTE executionType needs a ComputeTask, or its
Class, in the message body, or the taskName option");
} else {
throw new RuntimeCamelException(
String.format("Ignite Compute endpoint with EXECUTE
executionType is only "
@@ -173,8 +186,9 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
}
}
- private void doRun(final Exchange exchange, IgniteCompute compute) throws
NoTypeConversionAvailableException {
- Object job = exchange.getIn().getBody();
+ private void doRun(final Exchange exchange, IgniteCompute compute)
+ throws NoTypeConversionAvailableException, InvalidPayloadException
{
+ Object job = mandatoryJob(exchange, IgniteComputeExecutionType.RUN,
"an IgniteRunnable, or a Collection of them");
if (Collection.class.isAssignableFrom(job.getClass())) {
Collection<?> col = (Collection<?>) job;
@@ -196,7 +210,8 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
}
@SuppressWarnings("unchecked")
- private <T, R1, R2> void doApply(final Exchange exchange, IgniteCompute
compute) {
+ private <T, R1, R2> void doApply(final Exchange exchange, IgniteCompute
compute) throws InvalidPayloadException {
+ Object body = mandatoryJob(exchange, IgniteComputeExecutionType.APPLY,
"an IgniteClosure");
IgniteClosure<T, R1> job =
exchange.getIn().getBody(IgniteClosure.class);
T params = (T)
exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_PARAMS);
@@ -204,7 +219,7 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
throw new RuntimeCamelException(
String.format("Ignite Compute endpoint with APPLY
executionType is only "
+ "supported for IgniteClosure payloads with
parameters. The payload type was: %s.",
- exchange.getIn().getBody().getClass().getName()));
+ body.getClass().getName()));
}
IgniteReducer<R1, R2> reducer =
exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_REDUCER,
IgniteReducer.class);
@@ -222,7 +237,8 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
}
@SuppressWarnings("unchecked")
- private void doAffinityCall(final Exchange exchange, IgniteCompute
compute) {
+ private void doAffinityCall(final Exchange exchange, IgniteCompute
compute) throws InvalidPayloadException {
+ Object body = mandatoryJob(exchange,
IgniteComputeExecutionType.AFFINITY_CALL, "an IgniteCallable");
IgniteCallable<Object> job =
exchange.getIn().getBody(IgniteCallable.class);
String affinityCache =
exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_AFFINITY_CACHE_NAME,
String.class);
Object affinityKey =
exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_AFFINITY_KEY,
Object.class);
@@ -232,13 +248,14 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
String.format(
"Ignite Compute endpoint with AFFINITY_CALL
executionType is only "
+ "supported for IgniteCallable payloads,
along with an affinity cache and key. The payload type was: %s.",
- exchange.getIn().getBody().getClass().getName()));
+ body.getClass().getName()));
}
compute.affinityCall(affinityCache, affinityKey, job);
}
- private void doAffinityRun(final Exchange exchange, IgniteCompute compute)
{
+ private void doAffinityRun(final Exchange exchange, IgniteCompute compute)
throws InvalidPayloadException {
+ Object body = mandatoryJob(exchange,
IgniteComputeExecutionType.AFFINITY_RUN, "an IgniteRunnable");
IgniteRunnable job = exchange.getIn().getBody(IgniteRunnable.class);
String affinityCache =
exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_AFFINITY_CACHE_NAME,
String.class);
Object affinityKey =
exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_AFFINITY_KEY,
Object.class);
@@ -248,12 +265,27 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
String.format(
"Ignite Compute endpoint with AFFINITY_RUN
executionType is only "
+ "supported for IgniteRunnable payloads,
along with an affinity cache and key. The payload type was: %s.",
- exchange.getIn().getBody().getClass().getName()));
+ body.getClass().getName()));
}
compute.affinityRun(affinityCache, affinityKey, job);
}
+ /**
+ * The job in the body, or an {@link InvalidPayloadException} that says
what the execution type needs when there is
+ * no body.
+ */
+ private static Object mandatoryJob(Exchange exchange,
IgniteComputeExecutionType executionType, String job)
+ throws InvalidPayloadException {
+ Object body = exchange.getIn().getBody();
+ if (body == null) {
+ throw new InvalidPayloadException(
+ exchange, Object.class, exchange.getIn(),
+ "the " + executionType + " executionType needs " + job + "
in the message body");
+ }
+ return body;
+ }
+
private IgniteComputeExecutionType executionTypeFor(Exchange exchange) {
return
exchange.getIn().getHeader(IgniteConstants.IGNITE_COMPUTE_EXECUTION_TYPE,
endpoint.getExecutionType(),
IgniteComputeExecutionType.class);
@@ -264,11 +296,13 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
private Exchange exchange;
private AsyncCallback callback;
+ private boolean propagateIncomingBody;
- private static IgniteInCamelClosure create(Exchange exchange,
AsyncCallback callback) {
+ private static IgniteInCamelClosure create(Exchange exchange,
AsyncCallback callback, boolean propagateIncomingBody) {
IgniteInCamelClosure answer = new IgniteInCamelClosure();
answer.exchange = exchange;
answer.callback = callback;
+ answer.propagateIncomingBody = propagateIncomingBody;
return answer;
}
@@ -288,7 +322,8 @@ public class IgniteComputeProducer extends
DefaultAsyncProducer {
return;
}
- exchange.getOut().setBody(result);
+ // a job without a result keeps the incoming body when
propagateIncomingBodyIfNoReturnValue is true
+ out.setBody(propagateIncomingBody ? in.getBody() : result);
callback.done(false);
}
}
diff --git
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenEndpoint.java
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenEndpoint.java
index 0e739c296af6..efc0dca2a144 100644
---
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenEndpoint.java
+++
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenEndpoint.java
@@ -53,9 +53,9 @@ import static
org.apache.camel.component.ignite.IgniteConstants.SCHEME_IDGEN;
@BodyType.When(option = "operation",
value = "GET_AND_ADD",
carriers = {
Carrier.VALUE, Carrier.TEXT }) },
description = "ADD_AND_GET and GET_AND_ADD
add the body to the sequence: a number, or text"
- + " of one; no body fails. The
other operations do not use the body, but text"
- + " that is not a number still
fails. The operation is the operation option or"
- + " the
CamelIgniteIdGenOperation header"),
+ + " of one; no body fails. The
other operations do not use the body. The"
+ + " operation is the operation
option or the CamelIgniteIdGenOperation header;"
+ + " with neither the producer
fails"),
producerOut = @BodyType(value = Carrier.VALUE, javaType =
"java.lang.Long",
description = "The value of the sequence,
as a Long"))
public class IgniteIdGenEndpoint extends AbstractIgniteEndpoint {
diff --git
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenProducer.java
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenProducer.java
index 32a4f40edde2..d0141db84f82 100644
---
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenProducer.java
+++
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/idgen/IgniteIdGenProducer.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.ignite.idgen;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
import org.apache.camel.component.ignite.IgniteConstants;
import org.apache.camel.support.DefaultAsyncProducer;
@@ -41,12 +42,26 @@ public class IgniteIdGenProducer extends
DefaultAsyncProducer {
@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
Message in = exchange.getIn();
+ IgniteIdGenOperation operation = idGenOperationFor(exchange);
+
+ // only the operations that add to the sequence use the body
+ long id = 0;
+ if (operation == IgniteIdGenOperation.ADD_AND_GET || operation ==
IgniteIdGenOperation.GET_AND_ADD) {
+ Long number = in.getBody(Long.class);
+ if (number == null) {
+ exchange.setException(new InvalidPayloadException(
+ exchange, Long.class, in,
+ "the " + operation + " operation needs the number to
add in the message body"));
+ callback.done(true);
+ return true;
+ }
+ id = number;
+ }
+
Message out = exchange.getOut();
MessageHelper.copyHeaders(in, out, true);
- Long id = in.getBody(Long.class);
-
- switch (idGenOperationFor(exchange)) {
+ switch (operation) {
case ADD_AND_GET:
out.setBody(atomicSeq.addAndGet(id));
@@ -79,8 +94,14 @@ public class IgniteIdGenProducer extends
DefaultAsyncProducer {
}
private IgniteIdGenOperation idGenOperationFor(Exchange exchange) {
- return
exchange.getIn().getHeader(IgniteConstants.IGNITE_IDGEN_OPERATION,
endpoint.getOperation(),
- IgniteIdGenOperation.class);
+ IgniteIdGenOperation operation =
exchange.getIn().getHeader(IgniteConstants.IGNITE_IDGEN_OPERATION,
+ endpoint.getOperation(), IgniteIdGenOperation.class);
+ if (operation == null) {
+ throw new IllegalArgumentException(
+ "No operation specified. Set the operation via the
operation option of the endpoint or the header "
+ +
IgniteConstants.IGNITE_IDGEN_OPERATION);
+ }
+ return operation;
}
}
diff --git
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueEndpoint.java
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueEndpoint.java
index 3a896246cba9..0b4d9c51b8f1 100644
---
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueEndpoint.java
+++
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueEndpoint.java
@@ -50,10 +50,10 @@ import static
org.apache.camel.component.ignite.IgniteConstants.SCHEME_QUEUE;
@BodyTypes(producerIn = @BodyType(value = Carrier.ANY,
description = "ADD, OFFER, PUT, CONTAINS,
REMOVE and RETAIN_ALL use the body as the"
+ " element, and a Collection
as several elements unless"
- + "
treatCollectionsAsCacheObjects=true; no body fails ADD, CONTAINS, REMOVE"
- + " and RETAIN_ALL. DRAIN
drains into a Collection body, or a new List. The"
- + " other operations do not
use the body. The operation is the operation option"
- + " or the
CamelIgniteQueueOperation header; with neither the producer fails"),
+ + "
treatCollectionsAsCacheObjects=true; no body fails them. DRAIN drains into"
+ + " a Collection body, or a
new List. The other operations do not use the body."
+ + " The operation is the
operation option or the CamelIgniteQueueOperation"
+ + " header; with neither the
producer fails"),
producerOut = @BodyType(value = Carrier.UNKNOWN,
dependsOn =
"propagateIncomingBodyIfNoReturnValue",
byOption = {
diff --git
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueProducer.java
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueProducer.java
index 8b647e8f9950..f797099b1d2b 100644
---
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueProducer.java
+++
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/queue/IgniteQueueProducer.java
@@ -19,11 +19,14 @@ package org.apache.camel.component.ignite.queue;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.Iterator;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
import org.apache.camel.component.ignite.IgniteConstants;
import org.apache.camel.support.DefaultAsyncProducer;
@@ -32,6 +35,13 @@ import org.apache.ignite.IgniteQueue;
public class IgniteQueueProducer extends DefaultAsyncProducer {
+ /**
+ * The operations that use the body as the element (or the elements) to
add, find or remove.
+ */
+ private static final Set<IgniteQueueOperation> ELEMENT_OPERATIONS =
EnumSet.of(IgniteQueueOperation.ADD,
+ IgniteQueueOperation.CONTAINS, IgniteQueueOperation.REMOVE,
IgniteQueueOperation.RETAIN_ALL,
+ IgniteQueueOperation.OFFER, IgniteQueueOperation.PUT);
+
private IgniteQueueEndpoint endpoint;
private IgniteQueue<Object> queue;
@@ -45,13 +55,22 @@ public class IgniteQueueProducer extends
DefaultAsyncProducer {
@SuppressWarnings("unchecked")
public boolean process(Exchange exchange, AsyncCallback callback) {
Message in = exchange.getIn();
+ IgniteQueueOperation operation = queueOperationFor(exchange);
+
+ Object body = in.getBody();
+ if (body == null && ELEMENT_OPERATIONS.contains(operation)) {
+ exchange.setException(new InvalidPayloadException(
+ exchange, Object.class, in, "the " + operation + "
operation needs the element in the message body"));
+ callback.done(true);
+ return true;
+ }
+
Message out = exchange.getOut();
MessageHelper.copyHeaders(exchange.getIn(), out, true);
- Object body = in.getBody();
Long millis;
- switch (queueOperationFor(exchange)) {
+ switch (operation) {
case ADD:
if (Collection.class.isAssignableFrom(body.getClass()) &&
!endpoint.isTreatCollectionsAsCacheObjects()) {
@@ -165,8 +184,14 @@ public class IgniteQueueProducer extends
DefaultAsyncProducer {
}
private IgniteQueueOperation queueOperationFor(Exchange exchange) {
- return
exchange.getIn().getHeader(IgniteConstants.IGNITE_QUEUE_OPERATION,
endpoint.getOperation(),
- IgniteQueueOperation.class);
+ IgniteQueueOperation operation =
exchange.getIn().getHeader(IgniteConstants.IGNITE_QUEUE_OPERATION,
+ endpoint.getOperation(), IgniteQueueOperation.class);
+ if (operation == null) {
+ throw new IllegalArgumentException(
+ "No operation specified. Set the operation via the
operation option of the endpoint or the header "
+ +
IgniteConstants.IGNITE_QUEUE_OPERATION);
+ }
+ return operation;
}
}
diff --git
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetProducer.java
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetProducer.java
index e54ee847299c..ba55614c1ac5 100644
---
a/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetProducer.java
+++
b/components/camel-ignite/src/main/java/org/apache/camel/component/ignite/set/IgniteSetProducer.java
@@ -18,10 +18,13 @@ package org.apache.camel.component.ignite.set;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.Iterator;
+import java.util.Set;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
import org.apache.camel.component.ignite.IgniteConstants;
import org.apache.camel.support.DefaultAsyncProducer;
@@ -33,6 +36,12 @@ import org.apache.ignite.IgniteSet;
*/
public class IgniteSetProducer extends DefaultAsyncProducer {
+ /**
+ * The operations that use the body as the element (or the elements) to
add, find or remove.
+ */
+ private static final Set<IgniteSetOperation> ELEMENT_OPERATIONS =
EnumSet.of(IgniteSetOperation.ADD,
+ IgniteSetOperation.CONTAINS, IgniteSetOperation.REMOVE,
IgniteSetOperation.RETAIN_ALL);
+
private IgniteSetEndpoint endpoint;
private IgniteSet<Object> set;
@@ -46,12 +55,20 @@ public class IgniteSetProducer extends DefaultAsyncProducer
{
@SuppressWarnings("unchecked")
public boolean process(Exchange exchange, AsyncCallback callback) {
Message in = exchange.getIn();
- Message out = exchange.getOut();
- MessageHelper.copyHeaders(exchange.getIn(), out, true);
+ IgniteSetOperation operation = setOperationFor(exchange);
Object body = in.getBody();
+ if (body == null && ELEMENT_OPERATIONS.contains(operation)) {
+ exchange.setException(new InvalidPayloadException(
+ exchange, Object.class, in, "the " + operation + "
operation needs the element in the message body"));
+ callback.done(true);
+ return true;
+ }
- switch (setOperationFor(exchange)) {
+ Message out = exchange.getOut();
+ MessageHelper.copyHeaders(exchange.getIn(), out, true);
+
+ switch (operation) {
case ADD:
if (Collection.class.isAssignableFrom(body.getClass()) &&
!endpoint.isTreatCollectionsAsCacheObjects()) {
@@ -115,8 +132,14 @@ public class IgniteSetProducer extends
DefaultAsyncProducer {
}
private IgniteSetOperation setOperationFor(Exchange exchange) {
- return
exchange.getIn().getHeader(IgniteConstants.IGNITE_SETS_OPERATION,
endpoint.getOperation(),
- IgniteSetOperation.class);
+ IgniteSetOperation operation =
exchange.getIn().getHeader(IgniteConstants.IGNITE_SETS_OPERATION,
+ endpoint.getOperation(), IgniteSetOperation.class);
+ if (operation == null) {
+ throw new IllegalArgumentException(
+ "No operation specified. Set the operation via the
operation option of the endpoint or the header "
+ +
IgniteConstants.IGNITE_SETS_OPERATION);
+ }
+ return operation;
}
}
diff --git
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheContinuousQueryTest.java
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheContinuousQueryTest.java
index c6dc244743e0..8db2e3ca8be1 100644
---
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheContinuousQueryTest.java
+++
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheContinuousQueryTest.java
@@ -122,6 +122,27 @@ public class IgniteCacheContinuousQueryTest extends
AbstractIgniteTest implement
MockEndpoint.assertIsSatisfied(context);
}
+ @Test
+ void testContinuousQueryFireExistingEntriesSendsTheValueWithTheKey()
throws Exception {
+ getMockEndpoint("mock:test2").expectedMessageCount(50);
+
+ Map<Integer, Person> persons = createPersons(1, 100);
+ IgniteCache<Integer, Person> cache =
ignite().getOrCreateCache("testcontinuous1");
+ cache.putAll(persons);
+
+
context.getRouteController().startRoute("continuousQuery.fireExistingEntries");
+
+ MockEndpoint.assertIsSatisfied(context);
+
+ // the existing entries arrive like the events: the value as the body,
and the key in a header
+ for (Exchange exchange : getMockEndpoint("mock:test2").getExchanges())
{
+ Integer key =
exchange.getIn().getHeader(IgniteConstants.IGNITE_CACHE_KEY, Integer.class);
+ Assertions.assertThat(key).isNotNull().isGreaterThan(50);
+
Assertions.assertThat(exchange.getIn().getBody()).isEqualTo(persons.get(key));
+
Assertions.assertThat(exchange.getIn().getHeader(IgniteConstants.IGNITE_CACHE_NAME)).isEqualTo("testcontinuous1");
+ }
+ }
+
@Test
public void
testContinuousQueryFireExistingEntriesWithQueryAndRemoteFilter() throws
Exception {
getMockEndpoint("mock:test3").expectedMessageCount(50);
diff --git
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheTest.java
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheTest.java
index b075138e8f32..be86315c66a9 100644
---
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheTest.java
+++
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteCacheTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.ignite;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -25,6 +26,8 @@ import javax.cache.Cache.Entry;
import com.google.common.collect.ImmutableMap;
import org.apache.camel.CamelException;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.RuntimeCamelException;
import org.apache.camel.component.ignite.cache.IgniteCacheComponent;
import org.apache.camel.component.ignite.cache.IgniteCacheOperation;
import org.apache.camel.util.ObjectHelper;
@@ -258,6 +261,71 @@ public class IgniteCacheTest extends AbstractIgniteTest {
Assertions.assertThat(result).isNull();
}
+ @Test
+ void testPutMapWithKeyStoresTheMapAsOneValue() {
+ Map<String, String> map = new HashMap<>();
+ map.put("abcd", "1234");
+ map.put("efgh", "5678");
+
+ Object result = template.requestBodyAndHeader("ignite-cache:" +
resourceUid + "?operation=PUT", map,
+ IgniteConstants.IGNITE_CACHE_KEY, "orders");
+
+
Assertions.assertThat(ignite().cache(resourceUid).size(CachePeekMode.ALL)).isEqualTo(1);
+
Assertions.assertThat(ignite().cache(resourceUid).get("orders")).isEqualTo(map);
+ Assertions.assertThat(result).isEqualTo(map);
+ }
+
+ @Test
+ void testPutMapTreatCollectionsAsCacheObjectsStoresTheMapAsOneValue() {
+ Map<String, String> map = new HashMap<>();
+ map.put("abcd", "1234");
+ map.put("efgh", "5678");
+
+ template.requestBodyAndHeader("ignite-cache:" + resourceUid +
"?operation=PUT&treatCollectionsAsCacheObjects=true",
+ map, IgniteConstants.IGNITE_CACHE_KEY, "orders");
+
+
Assertions.assertThat(ignite().cache(resourceUid).size(CachePeekMode.ALL)).isEqualTo(1);
+
Assertions.assertThat(ignite().cache(resourceUid).get("orders")).isEqualTo(map);
+
+ // without a key there is no entry to store the Map in
+ Assertions
+ .assertThatThrownBy(() -> template.requestBody(
+ "ignite-cache:" + resourceUid +
"?operation=PUT&treatCollectionsAsCacheObjects=true", map))
+ .isInstanceOf(CamelExecutionException.class)
+ .cause().isInstanceOf(RuntimeCamelException.class)
+ .hasMessageContaining(IgniteConstants.IGNITE_CACHE_KEY);
+
Assertions.assertThat(ignite().cache(resourceUid).size(CachePeekMode.ALL)).isEqualTo(1);
+ }
+
+ @Test
+ void testPutMapEntriesPropagatesTheBody() {
+ Map<String, String> map = new HashMap<>();
+ map.put("abcd", "1234");
+ map.put("efgh", "5678");
+
+ Object result = template.requestBody("ignite-cache:" + resourceUid +
"?operation=PUT", map);
+
+
Assertions.assertThat(ignite().cache(resourceUid).size(CachePeekMode.ALL)).isEqualTo(2);
+ Assertions.assertThat(result).isEqualTo(map);
+
+ result = template.requestBody(
+ "ignite-cache:" + resourceUid +
"?operation=PUT&propagateIncomingBodyIfNoReturnValue=false", map);
+
+
Assertions.assertThat(ignite().cache(resourceUid).size(CachePeekMode.ALL)).isEqualTo(2);
+ Assertions.assertThat(result).isNull();
+ }
+
+ @Test
+ void testNoOperationFails() {
+ Assertions
+ .assertThatThrownBy(() ->
template.requestBodyAndHeader("ignite-cache:" + resourceUid, "1234",
+ IgniteConstants.IGNITE_CACHE_KEY, "abcd"))
+ .isInstanceOf(CamelExecutionException.class)
+ .cause().isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("operation option")
+ .hasMessageContaining(IgniteConstants.IGNITE_CACHE_OPERATION);
+ }
+
@AfterEach
public void deleteCaches() {
IgniteCache<?, ?> cache = ignite().cache(resourceUid);
diff --git
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteComputeTest.java
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteComputeTest.java
index b1a2694566a2..c17a31dafd77 100644
---
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteComputeTest.java
+++
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteComputeTest.java
@@ -26,8 +26,11 @@ import java.util.concurrent.TimeUnit;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.component.ignite.compute.IgniteComputeComponent;
+import org.apache.camel.component.ignite.compute.IgniteComputeExecutionType;
import org.apache.camel.util.ObjectHelper;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCluster;
@@ -103,21 +106,71 @@ public class IgniteComputeTest extends AbstractIgniteTest
{
public void testRun() {
TestIgniteComputeResources.COUNTER.set(0);
- // Single Runnable.
+ // Single Runnable: there is no result, so the body is kept.
Object result = template.requestBody("ignite-compute:" + resourceUid +
"?executionType=RUN",
TestIgniteComputeResources.TEST_RUNNABLE_COUNTER,
Object.class);
- Assertions.assertThat(result).isNull();
+
Assertions.assertThat(result).isSameAs(TestIgniteComputeResources.TEST_RUNNABLE_COUNTER);
Assertions.assertThat(TestIgniteComputeResources.COUNTER.get()).isEqualTo(1);
// Multiple Runnables.
Object[] runnables = new Object[5];
Arrays.fill(runnables,
TestIgniteComputeResources.TEST_RUNNABLE_COUNTER);
- result = template.requestBody("ignite-compute:" + resourceUid +
"?executionType=RUN", Lists.newArrayList(runnables),
- Collection.class);
- Assertions.assertThat(result).isNull();
+ List<Object> jobs = Lists.newArrayList(runnables);
+ result = template.requestBody("ignite-compute:" + resourceUid +
"?executionType=RUN", jobs, Collection.class);
+ Assertions.assertThat(result).isEqualTo(jobs);
Assertions.assertThat(TestIgniteComputeResources.COUNTER.get()).isEqualTo(6);
}
+ @Test
+ void testRunWithoutPropagateIncomingBodyLeavesNoBody() {
+ TestIgniteComputeResources.COUNTER.set(0);
+
+ Object result = template.requestBody(
+ "ignite-compute:" + resourceUid +
"?executionType=RUN&propagateIncomingBodyIfNoReturnValue=false",
+ TestIgniteComputeResources.TEST_RUNNABLE_COUNTER,
Object.class);
+ Assertions.assertThat(result).isNull();
+
Assertions.assertThat(TestIgniteComputeResources.COUNTER.get()).isEqualTo(1);
+ }
+
+ @Test
+ void testAffinityRunKeepsTheBody() {
+ TestIgniteComputeResources.COUNTER.set(0);
+ ignite().getOrCreateCache(resourceUid);
+
+ Map<String, Object> headers =
Map.of(IgniteConstants.IGNITE_COMPUTE_AFFINITY_CACHE_NAME, resourceUid,
+ IgniteConstants.IGNITE_COMPUTE_AFFINITY_KEY, "key");
+ Object result = template.requestBodyAndHeaders("ignite-compute:" +
resourceUid + "?executionType=AFFINITY_RUN",
+ TestIgniteComputeResources.TEST_RUNNABLE_COUNTER, headers,
Object.class);
+
Assertions.assertThat(result).isSameAs(TestIgniteComputeResources.TEST_RUNNABLE_COUNTER);
+
Assertions.assertThat(TestIgniteComputeResources.COUNTER.get()).isEqualTo(1);
+ }
+
+ @Test
+ void testNoBodyFails() {
+ for (IgniteComputeExecutionType executionType :
IgniteComputeExecutionType.values()) {
+ Assertions
+ .assertThatThrownBy(() -> template.requestBody(
+ "ignite-compute:" + resourceUid +
"?executionType=" + executionType, (Object) null))
+ .as(executionType.name())
+ .isInstanceOf(CamelExecutionException.class)
+ .cause().isInstanceOf(InvalidPayloadException.class)
+ .hasMessageContaining("the " + executionType + "
executionType needs");
+ }
+ }
+
+ @Test
+ void testExecuteTaskNameWithoutBodyTakesTheParametersHeader() {
+ String uri = "ignite-compute:" + resourceUid +
"?executionType=EXECUTE&taskName="
+ + TestIgniteComputeResources.EchoTask.class.getName();
+
+ String result = template.requestBodyAndHeader(uri, null,
IgniteConstants.IGNITE_COMPUTE_PARAMS, "Camel", String.class);
+ Assertions.assertThat(result).isEqualTo("echo Camel");
+
+ // a body is the argument of the task
+ result = template.requestBodyAndHeader(uri, "body",
IgniteConstants.IGNITE_COMPUTE_PARAMS, "Camel", String.class);
+ Assertions.assertThat(result).isEqualTo("echo body");
+ }
+
@Test
@SuppressWarnings("unchecked")
public void testBroadcast() {
diff --git
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteIdGenTest.java
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteIdGenTest.java
index 1cb5db601f57..21e4a71b2812 100644
---
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteIdGenTest.java
+++
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteIdGenTest.java
@@ -16,6 +16,10 @@
*/
package org.apache.camel.component.ignite;
+import java.util.List;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.component.ignite.idgen.IgniteIdGenComponent;
import org.apache.camel.component.ignite.idgen.IgniteIdGenEndpoint;
import org.apache.camel.component.ignite.idgen.IgniteIdGenOperation;
@@ -109,6 +113,46 @@ public class IgniteIdGenTest extends AbstractIgniteTest {
Assertions.assertThat(endpoint.getBatchSize()).isEqualTo(100);
}
+ @Test
+ void testOperationsThatDoNotAddIgnoreTheBody() {
+ Assertions.assertThat(template.requestBody("ignite-idgen:" +
resourceUid + "?initialValue=0&operation=GET",
+ "not a number", Long.class)).isZero();
+ Assertions.assertThat(template.requestBody(
+ "ignite-idgen:" + resourceUid +
"?initialValue=0&operation=GET_AND_INCREMENT", "not a number", Long.class))
+ .isZero();
+ Assertions.assertThat(template.requestBody(
+ "ignite-idgen:" + resourceUid +
"?initialValue=0&operation=INCREMENT_AND_GET", "not a number", Long.class))
+ .isEqualTo(2);
+ }
+
+ @Test
+ void testAddWithoutBodyFails() {
+ for (IgniteIdGenOperation operation :
List.of(IgniteIdGenOperation.ADD_AND_GET, IgniteIdGenOperation.GET_AND_ADD)) {
+ Assertions
+ .assertThatThrownBy(() -> template.requestBody(
+ "ignite-idgen:" + resourceUid +
"?initialValue=0&operation=" + operation, (Object) null))
+ .isInstanceOf(CamelExecutionException.class)
+ .cause().isInstanceOf(InvalidPayloadException.class)
+ .hasMessageContaining("the " + operation + " operation
needs the number to add");
+ }
+
+ // the sequence is unchanged
+ Assertions.assertThat(
+ template.requestBody("ignite-idgen:" + resourceUid +
"?initialValue=0&operation=GET", null, Long.class))
+ .isZero();
+ }
+
+ @Test
+ void testNoOperationFails() {
+ Assertions
+ .assertThatThrownBy(
+ () -> template.requestBody("ignite-idgen:" +
resourceUid + "?initialValue=0", (Object) null))
+ .isInstanceOf(CamelExecutionException.class)
+ .cause().isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("operation option")
+ .hasMessageContaining(IgniteConstants.IGNITE_IDGEN_OPERATION);
+ }
+
@AfterEach
public void deleteSets() {
IgniteAtomicSequence seq = ignite().atomicSequence(resourceUid, 0,
false);
diff --git
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteQueueTest.java
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteQueueTest.java
index e166b2d69491..fda1bd47ee8b 100644
---
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteQueueTest.java
+++
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteQueueTest.java
@@ -28,6 +28,8 @@ import com.google.common.base.Stopwatch;
import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.component.ignite.queue.IgniteQueueComponent;
import org.apache.camel.component.ignite.queue.IgniteQueueEndpoint;
import org.apache.camel.component.ignite.queue.IgniteQueueOperation;
@@ -275,6 +277,33 @@ public class IgniteQueueTest extends AbstractIgniteTest {
}
+ @Test
+ void testOperationsWithAnElementFailWithoutBody() {
+ for (IgniteQueueOperation operation :
List.of(IgniteQueueOperation.ADD, IgniteQueueOperation.CONTAINS,
+ IgniteQueueOperation.REMOVE, IgniteQueueOperation.RETAIN_ALL,
IgniteQueueOperation.OFFER,
+ IgniteQueueOperation.PUT)) {
+ Assertions
+ .assertThatThrownBy(() -> template.requestBody(
+ "ignite-queue:" + resourceUid + "?operation=" +
operation, (Object) null))
+ .as(operation.name())
+ .isInstanceOf(CamelExecutionException.class)
+ .cause().isInstanceOf(InvalidPayloadException.class)
+ .hasMessageContaining("the " + operation + " operation
needs the element");
+ }
+
+ Assertions.assertThat(ignite().queue(resourceUid, 0, new
CollectionConfiguration())).isEmpty();
+ }
+
+ @Test
+ void testNoOperationFails() {
+ Assertions
+ .assertThatThrownBy(() -> template.requestBody("ignite-queue:"
+ resourceUid, "hello"))
+ .isInstanceOf(CamelExecutionException.class)
+ .cause().isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("operation option")
+ .hasMessageContaining(IgniteConstants.IGNITE_QUEUE_OPERATION);
+ }
+
@AfterEach
public void deleteQueue() {
ignite().queue(resourceUid, 0, null).close();
diff --git
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteSetTest.java
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteSetTest.java
index 71005de3b981..0ba8dacbc038 100644
---
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteSetTest.java
+++
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/IgniteSetTest.java
@@ -23,6 +23,8 @@ import java.util.Set;
import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.InvalidPayloadException;
import org.apache.camel.component.ignite.set.IgniteSetComponent;
import org.apache.camel.component.ignite.set.IgniteSetEndpoint;
import org.apache.camel.component.ignite.set.IgniteSetOperation;
@@ -174,6 +176,32 @@ public class IgniteSetTest extends AbstractIgniteTest {
Assertions.assertThat(igniteEndpoint.getConfiguration()).isEqualTo(configuration);
}
+ @Test
+ void testOperationsWithAnElementFailWithoutBody() {
+ for (IgniteSetOperation operation : List.of(IgniteSetOperation.ADD,
IgniteSetOperation.CONTAINS,
+ IgniteSetOperation.REMOVE, IgniteSetOperation.RETAIN_ALL)) {
+ Assertions
+ .assertThatThrownBy(() -> template.requestBody(
+ "ignite-set:" + resourceUid + "?operation=" +
operation, (Object) null))
+ .as(operation.name())
+ .isInstanceOf(CamelExecutionException.class)
+ .cause().isInstanceOf(InvalidPayloadException.class)
+ .hasMessageContaining("the " + operation + " operation
needs the element");
+ }
+
+ Assertions.assertThat(ignite().set(resourceUid, new
CollectionConfiguration())).isEmpty();
+ }
+
+ @Test
+ void testNoOperationFails() {
+ Assertions
+ .assertThatThrownBy(() -> template.requestBody("ignite-set:" +
resourceUid, "hello"))
+ .isInstanceOf(CamelExecutionException.class)
+ .cause().isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("operation option")
+ .hasMessageContaining(IgniteConstants.IGNITE_SETS_OPERATION);
+ }
+
@AfterEach
public void deleteSet() {
ignite().set(resourceUid, null).close();
diff --git
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/TestIgniteComputeResources.java
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/TestIgniteComputeResources.java
index dfb7da743884..b2bef6a722c7 100644
---
a/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/TestIgniteComputeResources.java
+++
b/components/camel-ignite/src/test/java/org/apache/camel/component/ignite/TestIgniteComputeResources.java
@@ -27,6 +27,7 @@ import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
import org.apache.ignite.IgniteException;
import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobAdapter;
import org.apache.ignite.compute.ComputeJobResult;
import org.apache.ignite.compute.ComputeTask;
import org.apache.ignite.compute.ComputeTaskSplitAdapter;
@@ -146,4 +147,29 @@ public final class TestIgniteComputeResources {
private TestIgniteComputeResources() {
}
+
+ /**
+ * A task with one job, which gives back its argument after "echo ".
Ignite deploys it by its class name, which the
+ * taskName option of the endpoint gives.
+ */
+ public static class EchoTask extends ComputeTaskSplitAdapter<Object,
String> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ protected Collection<? extends ComputeJob> split(int gridSize, Object
arg) throws IgniteException {
+ return Collections.singleton(new ComputeJobAdapter(arg) {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Object execute() throws IgniteException {
+ return "echo " + argument(0);
+ }
+ });
+ }
+
+ @Override
+ public String reduce(List<ComputeJobResult> results) throws
IgniteException {
+ return results.get(0).getData();
+ }
+ }
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 7e73cbe50c42..fcde13086ce7 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -4942,6 +4942,31 @@ conversion error. Previously it gave an empty SOAP body.
A reply body that is an
The `spring-ws` producer now sends a `File` body, the file of a `file` or
`ftp` consumer, and an `InputStream` body
(also when stream caching is off) as XML. Previously these failed with an
`InvalidPayloadException`.
+=== camel-ignite - cache PUT of a Map, the existing query results, RUN bodies
and missing operations
+
+With `fireExistingQueryResults=true`, the `ignite-cache` consumer now sends
each entry that the initial query finds
+with the value as the body and the key in the `CamelIgniteCacheKey` header,
like the events that follow. Before, the
+body was the `javax.cache.Cache.Entry` and the key header was not set.
+
+A PUT of a `Map` body on `ignite-cache` with the `CamelIgniteCacheKey` header
now stores the `Map` as the value of that
+key. Before, the header was ignored and all the entries of the `Map` were put.
The same holds with
+`treatCollectionsAsCacheObjects=true`, where a PUT of a `Map` without the
header now fails, as there is no key to store
+it under. A PUT of a `Map` without the header still puts all its entries, and
now keeps the body when
+`propagateIncomingBodyIfNoReturnValue` is `true` (the default). Before, the
body was `null` after such a PUT.
+
+The `RUN` and `AFFINITY_RUN` execution types of `ignite-compute` have no
result, and now keep the body (the job) when
+`propagateIncomingBodyIfNoReturnValue` is `true` (the default). Before, the
body was `null`. Set
+`propagateIncomingBodyIfNoReturnValue=false` to keep the previous behaviour.
+
+The `ignite-cache`, `ignite-idgen`, `ignite-queue` and `ignite-set` producers
now fail with an
+`IllegalArgumentException` that names the `operation` option and the operation
header when neither sets the operation.
+A missing body now fails with an `InvalidPayloadException` that names what the
operation needs: the job for
+`ignite-compute` (except `EXECUTE` with the `taskName` option, which now runs
the task with the
+`CamelIgniteComputeParameters` header), the number to add for `ADD_AND_GET`
and `GET_AND_ADD` of `ignite-idgen`, and
+the element for `ADD`, `CONTAINS`, `REMOVE` and `RETAIN_ALL` of `ignite-queue`
and `ignite-set` and for `OFFER` and
+`PUT` of `ignite-queue`. Before, these failed with a `NullPointerException`.
The `GET`, `GET_AND_INCREMENT` and
+`INCREMENT_AND_GET` operations of `ignite-idgen` no longer fail on a body that
is not a number, as they do not use it.
+
=== camel-couchbase - the document id, the operation option and the bodies of
CCB_PUT
The `couchbase` producer now uses the `key` option as the document id when the
message has no `CamelCouchbaseId`
