You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2024/01/23 15:30:03 UTC
(camel) 13/19: CAMEL-19749: EIPs should make it easy to use together with variables.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch var
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 6a783e47ea2c13bab908719bf1efcd758af6804f
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue Jan 23 10:51:27 2024 +0100
CAMEL-19749: EIPs should make it easy to use together with variables.
---
.../apache/camel/catalog/models/pollEnrich.json | 17 +++---
.../apache/camel/catalog/schemas/camel-spring.xsd | 11 ++++
.../processor/SpringPollEnrichVariableTest.java | 29 ++++++++++
.../spring/processor/PollEnrichVariableTest.xml | 44 +++++++++++++++
.../org/apache/camel/model/pollEnrich.json | 17 +++---
.../apache/camel/model/PollEnrichDefinition.java | 22 ++++++++
.../org/apache/camel/processor/PollEnricher.java | 28 ++++++++++
.../apache/camel/reifier/PollEnrichReifier.java | 1 +
.../org/apache/camel/reifier/RouteReifier.java | 2 +-
.../java/org/apache/camel/reifier/SendReifier.java | 4 +-
.../org/apache/camel/reifier/ToDynamicReifier.java | 4 +-
.../org/apache/camel/reifier/WireTapReifier.java | 4 +-
.../camel/processor/PollEnrichVariableTest.java | 52 ++++++++++++++++++
.../management/mbean/ManagedPollEnricherMBean.java | 3 +
.../management/mbean/ManagedPollEnricher.java | 5 ++
.../java/org/apache/camel/xml/in/ModelParser.java | 1 +
.../java/org/apache/camel/xml/out/ModelWriter.java | 1 +
.../org/apache/camel/yaml/out/ModelWriter.java | 1 +
.../dsl/yaml/deserializers/ModelDeserializers.java | 8 ++-
.../generated/resources/schema/camelYamlDsl.json | 5 ++
.../camel/dsl/yaml/PollEnrichVariableTest.groovy | 64 ++++++++++++++++++++++
21 files changed, 299 insertions(+), 24 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/pollEnrich.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/pollEnrich.json
index e3ec0d66eb8..9f4fb8bafc5 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/pollEnrich.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/models/pollEnrich.json
@@ -16,13 +16,14 @@
"description": { "index": 1, "kind": "element", "displayName": "Description", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the description of this node" },
"disabled": { "index": 2, "kind": "attribute", "displayName": "Disabled", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to disable this EIP from the route during build time. Once an EIP has been disabled then it cannot be enabled later at runtime." },
"expression": { "index": 3, "kind": "expression", "displayName": "Expression", "required": true, "type": "object", "javaType": "org.apache.camel.model.language.ExpressionDefinition", "oneOf": [ "constant", "csimple", "datasonnet", "exchangeProperty", "groovy", "header", "hl7terser", "java", "joor", "jq", "js", "jsonpath", "language", "method", "mvel", "ognl", "python", "ref", "simple", "spel", "tokenize", "xpath", "xquery", "xtokenize" ], "deprecated": false, "autowired": false, "sec [...]
- "aggregationStrategy": { "index": 4, "kind": "attribute", "displayName": "Aggregation Strategy", "required": false, "type": "object", "javaType": "org.apache.camel.AggregationStrategy", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the AggregationStrategy to be used to merge the reply from the external service, into a single outgoing message. By default Camel will use the reply from the external service as outgoing message." },
- "aggregationStrategyMethodName": { "index": 5, "kind": "attribute", "displayName": "Aggregation Strategy Method Name", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "This option can be used to explicit declare the method name to use, when using POJOs as the AggregationStrategy." },
- "aggregationStrategyMethodAllowNull": { "index": 6, "kind": "attribute", "displayName": "Aggregation Strategy Method Allow Null", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "If this option is false then the aggregate method is not used if there was no data to enrich. If this option is true then null values is used as the oldExchange (when no data to enrich), when us [...]
- "aggregateOnException": { "index": 7, "kind": "attribute", "displayName": "Aggregate On Exception", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If this option is false then the aggregate method is not used if there was an exception thrown while trying to retrieve the data to enrich from the resource. Setting this option to true allows end us [...]
- "timeout": { "index": 8, "kind": "attribute", "displayName": "Timeout", "required": false, "type": "duration", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "-1", "description": "Timeout in millis when polling from the external service. The timeout has influence about the poll enrich behavior. It basically operations in three different modes: negative value - Waits until a message is available and then returns it. Warning th [...]
- "cacheSize": { "index": 9, "kind": "attribute", "displayName": "Cache Size", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the maximum size used by the org.apache.camel.spi.ConsumerCache which is used to cache and reuse consumers when uris are reused. Beware that when using dynamic endpoints then it affects how well the cache can be utilized. If each dynamic en [...]
- "ignoreInvalidEndpoint": { "index": 10, "kind": "attribute", "displayName": "Ignore Invalid Endpoint", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Ignore the invalidate endpoint exception when try to create a producer with that endpoint" },
- "autoStartComponents": { "index": 11, "kind": "attribute", "displayName": "Auto Start Components", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether to auto startup components when poll enricher is starting up." }
+ "variableReceive": { "index": 4, "kind": "attribute", "displayName": "Variable Receive", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "To use a variable to store the received message body (only body, not headers). This is handy for easy access to the received message body via variables. Important: When using receive variable then the received body is stored only in this variable and not o [...]
+ "aggregationStrategy": { "index": 5, "kind": "attribute", "displayName": "Aggregation Strategy", "required": false, "type": "object", "javaType": "org.apache.camel.AggregationStrategy", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the AggregationStrategy to be used to merge the reply from the external service, into a single outgoing message. By default Camel will use the reply from the external service as outgoing message." },
+ "aggregationStrategyMethodName": { "index": 6, "kind": "attribute", "displayName": "Aggregation Strategy Method Name", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "This option can be used to explicit declare the method name to use, when using POJOs as the AggregationStrategy." },
+ "aggregationStrategyMethodAllowNull": { "index": 7, "kind": "attribute", "displayName": "Aggregation Strategy Method Allow Null", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "If this option is false then the aggregate method is not used if there was no data to enrich. If this option is true then null values is used as the oldExchange (when no data to enrich), when us [...]
+ "aggregateOnException": { "index": 8, "kind": "attribute", "displayName": "Aggregate On Exception", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If this option is false then the aggregate method is not used if there was an exception thrown while trying to retrieve the data to enrich from the resource. Setting this option to true allows end us [...]
+ "timeout": { "index": 9, "kind": "attribute", "displayName": "Timeout", "required": false, "type": "duration", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "-1", "description": "Timeout in millis when polling from the external service. The timeout has influence about the poll enrich behavior. It basically operations in three different modes: negative value - Waits until a message is available and then returns it. Warning th [...]
+ "cacheSize": { "index": 10, "kind": "attribute", "displayName": "Cache Size", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the maximum size used by the org.apache.camel.spi.ConsumerCache which is used to cache and reuse consumers when uris are reused. Beware that when using dynamic endpoints then it affects how well the cache can be utilized. If each dynamic e [...]
+ "ignoreInvalidEndpoint": { "index": 11, "kind": "attribute", "displayName": "Ignore Invalid Endpoint", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Ignore the invalidate endpoint exception when try to create a producer with that endpoint" },
+ "autoStartComponents": { "index": 12, "kind": "attribute", "displayName": "Auto Start Components", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether to auto startup components when poll enricher is starting up." }
}
}
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
index 9360b3518d2..be712c5bf41 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd
@@ -11115,6 +11115,17 @@ Sets a reference to use for lookup the policy in the registry.
<xs:element ref="tns:xpath"/>
<xs:element ref="tns:xquery"/>
</xs:choice>
+ <xs:attribute name="variableReceive" type="xs:string">
+ <xs:annotation>
+ <xs:documentation xml:lang="en">
+<![CDATA[
+To use a variable to store the received message body (only body, not headers). This is handy for easy access to the
+received message body via variables. Important: When using receive variable then the received body is stored only in
+this variable and not on the current org.apache.camel.Message .
+]]>
+ </xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
<xs:attribute name="aggregationStrategy" type="xs:string">
<xs:annotation>
<xs:documentation xml:lang="en">
diff --git a/components/camel-spring-xml/src/test/java/org/apache/camel/spring/processor/SpringPollEnrichVariableTest.java b/components/camel-spring-xml/src/test/java/org/apache/camel/spring/processor/SpringPollEnrichVariableTest.java
new file mode 100644
index 00000000000..71278e79a04
--- /dev/null
+++ b/components/camel-spring-xml/src/test/java/org/apache/camel/spring/processor/SpringPollEnrichVariableTest.java
@@ -0,0 +1,29 @@
+/*
+ * 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.spring.processor;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.processor.PollEnrichVariableTest;
+
+import static org.apache.camel.spring.processor.SpringTestHelper.createSpringCamelContext;
+
+public class SpringPollEnrichVariableTest extends PollEnrichVariableTest {
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ return createSpringCamelContext(this, "org/apache/camel/spring/processor/PollEnrichVariableTest.xml");
+ }
+}
diff --git a/components/camel-spring-xml/src/test/resources/org/apache/camel/spring/processor/PollEnrichVariableTest.xml b/components/camel-spring-xml/src/test/resources/org/apache/camel/spring/processor/PollEnrichVariableTest.xml
new file mode 100644
index 00000000000..cbe82854d8c
--- /dev/null
+++ b/components/camel-spring-xml/src/test/resources/org/apache/camel/spring/processor/PollEnrichVariableTest.xml
@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<beans xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="
+ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+ http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
+ ">
+
+ <!-- START SNIPPET: example -->
+ <camelContext xmlns="http://camel.apache.org/schema/spring">
+ <jmxAgent id="jmx" disabled="true"/>
+ <route>
+ <from uri="direct:receive"/>
+ <pollEnrich timeout="1000" variableReceive="bye">
+ <constant>seda:foo</constant>
+ </pollEnrich>
+ <to uri="mock:after"/>
+ <setBody>
+ <simple>${variable:bye}</simple>
+ </setBody>
+ <to uri="mock:result"/>
+ </route>
+ </camelContext>
+ <!-- END SNIPPET: example -->
+
+</beans>
diff --git a/core/camel-core-model/src/generated/resources/org/apache/camel/model/pollEnrich.json b/core/camel-core-model/src/generated/resources/org/apache/camel/model/pollEnrich.json
index e3ec0d66eb8..9f4fb8bafc5 100644
--- a/core/camel-core-model/src/generated/resources/org/apache/camel/model/pollEnrich.json
+++ b/core/camel-core-model/src/generated/resources/org/apache/camel/model/pollEnrich.json
@@ -16,13 +16,14 @@
"description": { "index": 1, "kind": "element", "displayName": "Description", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the description of this node" },
"disabled": { "index": 2, "kind": "attribute", "displayName": "Disabled", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether to disable this EIP from the route during build time. Once an EIP has been disabled then it cannot be enabled later at runtime." },
"expression": { "index": 3, "kind": "expression", "displayName": "Expression", "required": true, "type": "object", "javaType": "org.apache.camel.model.language.ExpressionDefinition", "oneOf": [ "constant", "csimple", "datasonnet", "exchangeProperty", "groovy", "header", "hl7terser", "java", "joor", "jq", "js", "jsonpath", "language", "method", "mvel", "ognl", "python", "ref", "simple", "spel", "tokenize", "xpath", "xquery", "xtokenize" ], "deprecated": false, "autowired": false, "sec [...]
- "aggregationStrategy": { "index": 4, "kind": "attribute", "displayName": "Aggregation Strategy", "required": false, "type": "object", "javaType": "org.apache.camel.AggregationStrategy", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the AggregationStrategy to be used to merge the reply from the external service, into a single outgoing message. By default Camel will use the reply from the external service as outgoing message." },
- "aggregationStrategyMethodName": { "index": 5, "kind": "attribute", "displayName": "Aggregation Strategy Method Name", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "This option can be used to explicit declare the method name to use, when using POJOs as the AggregationStrategy." },
- "aggregationStrategyMethodAllowNull": { "index": 6, "kind": "attribute", "displayName": "Aggregation Strategy Method Allow Null", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "If this option is false then the aggregate method is not used if there was no data to enrich. If this option is true then null values is used as the oldExchange (when no data to enrich), when us [...]
- "aggregateOnException": { "index": 7, "kind": "attribute", "displayName": "Aggregate On Exception", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If this option is false then the aggregate method is not used if there was an exception thrown while trying to retrieve the data to enrich from the resource. Setting this option to true allows end us [...]
- "timeout": { "index": 8, "kind": "attribute", "displayName": "Timeout", "required": false, "type": "duration", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "-1", "description": "Timeout in millis when polling from the external service. The timeout has influence about the poll enrich behavior. It basically operations in three different modes: negative value - Waits until a message is available and then returns it. Warning th [...]
- "cacheSize": { "index": 9, "kind": "attribute", "displayName": "Cache Size", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the maximum size used by the org.apache.camel.spi.ConsumerCache which is used to cache and reuse consumers when uris are reused. Beware that when using dynamic endpoints then it affects how well the cache can be utilized. If each dynamic en [...]
- "ignoreInvalidEndpoint": { "index": 10, "kind": "attribute", "displayName": "Ignore Invalid Endpoint", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Ignore the invalidate endpoint exception when try to create a producer with that endpoint" },
- "autoStartComponents": { "index": 11, "kind": "attribute", "displayName": "Auto Start Components", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether to auto startup components when poll enricher is starting up." }
+ "variableReceive": { "index": 4, "kind": "attribute", "displayName": "Variable Receive", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "To use a variable to store the received message body (only body, not headers). This is handy for easy access to the received message body via variables. Important: When using receive variable then the received body is stored only in this variable and not o [...]
+ "aggregationStrategy": { "index": 5, "kind": "attribute", "displayName": "Aggregation Strategy", "required": false, "type": "object", "javaType": "org.apache.camel.AggregationStrategy", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the AggregationStrategy to be used to merge the reply from the external service, into a single outgoing message. By default Camel will use the reply from the external service as outgoing message." },
+ "aggregationStrategyMethodName": { "index": 6, "kind": "attribute", "displayName": "Aggregation Strategy Method Name", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "This option can be used to explicit declare the method name to use, when using POJOs as the AggregationStrategy." },
+ "aggregationStrategyMethodAllowNull": { "index": 7, "kind": "attribute", "displayName": "Aggregation Strategy Method Allow Null", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "If this option is false then the aggregate method is not used if there was no data to enrich. If this option is true then null values is used as the oldExchange (when no data to enrich), when us [...]
+ "aggregateOnException": { "index": 8, "kind": "attribute", "displayName": "Aggregate On Exception", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If this option is false then the aggregate method is not used if there was an exception thrown while trying to retrieve the data to enrich from the resource. Setting this option to true allows end us [...]
+ "timeout": { "index": 9, "kind": "attribute", "displayName": "Timeout", "required": false, "type": "duration", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "-1", "description": "Timeout in millis when polling from the external service. The timeout has influence about the poll enrich behavior. It basically operations in three different modes: negative value - Waits until a message is available and then returns it. Warning th [...]
+ "cacheSize": { "index": 10, "kind": "attribute", "displayName": "Cache Size", "label": "advanced", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "Sets the maximum size used by the org.apache.camel.spi.ConsumerCache which is used to cache and reuse consumers when uris are reused. Beware that when using dynamic endpoints then it affects how well the cache can be utilized. If each dynamic e [...]
+ "ignoreInvalidEndpoint": { "index": 11, "kind": "attribute", "displayName": "Ignore Invalid Endpoint", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Ignore the invalidate endpoint exception when try to create a producer with that endpoint" },
+ "autoStartComponents": { "index": 12, "kind": "attribute", "displayName": "Auto Start Components", "label": "advanced", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether to auto startup components when poll enricher is starting up." }
}
}
diff --git a/core/camel-core-model/src/main/java/org/apache/camel/model/PollEnrichDefinition.java b/core/camel-core-model/src/main/java/org/apache/camel/model/PollEnrichDefinition.java
index b61e9604c68..c471bdbb73d 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/model/PollEnrichDefinition.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/PollEnrichDefinition.java
@@ -39,6 +39,8 @@ public class PollEnrichDefinition extends ExpressionNode implements AggregationS
@XmlTransient
private AggregationStrategy aggregationStrategyBean;
+ @XmlAttribute
+ private String variableReceive;
@XmlAttribute
@Metadata(javaType = "org.apache.camel.AggregationStrategy")
private String aggregationStrategy;
@@ -130,6 +132,18 @@ public class PollEnrichDefinition extends ExpressionNode implements AggregationS
return this;
}
+ /**
+ * To use a variable to store the received message body (only body, not headers). This is handy for easy access to
+ * the received message body via variables.
+ *
+ * Important: When using receive variable then the received body is stored only in this variable and <b>not</b> on
+ * the current {@link org.apache.camel.Message}.
+ */
+ public PollEnrichDefinition variableReceive(String variableReceive) {
+ this.variableReceive = variableReceive;
+ return this;
+ }
+
/**
* Sets the AggregationStrategy to be used to merge the reply from the external service, into a single outgoing
* message. By default Camel will use the reply from the external service as outgoing message.
@@ -277,6 +291,14 @@ public class PollEnrichDefinition extends ExpressionNode implements AggregationS
this.timeout = timeout;
}
+ public String getVariableReceive() {
+ return variableReceive;
+ }
+
+ public void setVariableReceive(String variableReceive) {
+ this.variableReceive = variableReceive;
+ }
+
public String getAggregationStrategy() {
return aggregationStrategy;
}
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
index 109266a521b..ca029a6525a 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
@@ -68,6 +68,7 @@ public class PollEnricher extends AsyncProcessorSupport implements IdAware, Rout
private String id;
private String routeId;
private AggregationStrategy aggregationStrategy;
+ private String variableReceive;
private final Expression expression;
private final String uri;
private long timeout;
@@ -152,6 +153,14 @@ public class PollEnricher extends AsyncProcessorSupport implements IdAware, Rout
this.aggregationStrategy = aggregationStrategy;
}
+ public String getVariableReceive() {
+ return variableReceive;
+ }
+
+ public void setVariableReceive(String variableReceive) {
+ this.variableReceive = variableReceive;
+ }
+
public long getTimeout() {
return timeout;
}
@@ -308,6 +317,19 @@ public class PollEnricher extends AsyncProcessorSupport implements IdAware, Rout
cause = resourceExchange.getException();
}
+ // if we should store the received message body in a variable,
+ // then we need to preserve the original message body
+ Object originalBody = null;
+ if (variableReceive != null) {
+ try {
+ originalBody = exchange.getMessage().getBody();
+ } catch (Exception throwable) {
+ exchange.setException(throwable);
+ callback.done(true);
+ return true;
+ }
+ }
+
try {
if (!isAggregateOnException() && resourceExchange != null && resourceExchange.isFailed()) {
// copy resource exchange onto original exchange (preserving pattern)
@@ -321,6 +343,12 @@ public class PollEnricher extends AsyncProcessorSupport implements IdAware, Rout
// must catch any exception from aggregation
Exchange aggregatedExchange = aggregationStrategy.aggregate(exchange, resourceExchange);
if (aggregatedExchange != null) {
+ if (variableReceive != null) {
+ // result should be stored in variable instead of message body
+ Object value = aggregatedExchange.getMessage().getBody();
+ ExchangeHelper.setVariable(exchange, variableReceive, value);
+ aggregatedExchange.getMessage().setBody(originalBody);
+ }
// copy aggregation result onto original exchange (preserving pattern)
copyResultsPreservePattern(exchange, aggregatedExchange);
// handover any synchronization
diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/PollEnrichReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/PollEnrichReifier.java
index 4ddcbcc6ed8..2fb2f72a281 100644
--- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/PollEnrichReifier.java
+++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/PollEnrichReifier.java
@@ -67,6 +67,7 @@ public class PollEnrichReifier extends ProcessorReifier<PollEnrichDefinition> {
if (num != null) {
enricher.setCacheSize(num);
}
+ enricher.setVariableReceive(parseString(definition.getVariableReceive()));
enricher.setIgnoreInvalidEndpoint(parseBoolean(definition.getIgnoreInvalidEndpoint(), false));
enricher.setAggregateOnException(parseBoolean(definition.getAggregateOnException(), false));
if (definition.getAutoStartComponents() != null) {
diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RouteReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RouteReifier.java
index 97a8b686a02..d53870207a7 100644
--- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RouteReifier.java
+++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/RouteReifier.java
@@ -111,7 +111,7 @@ public class RouteReifier extends ProcessorReifier<RouteDefinition> {
route.setErrorHandlerFactory(definition.getErrorHandlerFactory());
// configure variable
- String variable = definition.getInput().getVariableReceive();
+ String variable = parseString(definition.getInput().getVariableReceive());
if (variable != null) {
// when using variable we need to turn on original message
route.setAllowUseOriginalMessage(true);
diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SendReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SendReifier.java
index 4eaa13a4f17..f01407d6f87 100644
--- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SendReifier.java
+++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/SendReifier.java
@@ -35,8 +35,8 @@ public class SendReifier extends ProcessorReifier<ToDefinition> {
@Override
public Processor createProcessor() throws Exception {
SendProcessor answer = new SendProcessor(resolveEndpoint(), parse(ExchangePattern.class, definition.getPattern()));
- answer.setVariableSend(definition.getVariableSend());
- answer.setVariableReceive(definition.getVariableReceive());
+ answer.setVariableSend(parseString(definition.getVariableSend()));
+ answer.setVariableReceive(parseString(definition.getVariableReceive()));
return answer;
}
diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/ToDynamicReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/ToDynamicReifier.java
index cfa004429fa..240d5e18e20 100644
--- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/ToDynamicReifier.java
+++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/ToDynamicReifier.java
@@ -56,8 +56,8 @@ public class ToDynamicReifier<T extends ToDynamicDefinition> extends ProcessorRe
SendDynamicProcessor processor = new SendDynamicProcessor(uri, exp);
processor.setCamelContext(camelContext);
processor.setPattern(parse(ExchangePattern.class, definition.getPattern()));
- processor.setVariableSend(definition.getVariableSend());
- processor.setVariableReceive(definition.getVariableReceive());
+ processor.setVariableSend(parseString(definition.getVariableSend()));
+ processor.setVariableReceive(parseString(definition.getVariableReceive()));
Integer num = parseInt(definition.getCacheSize());
if (num != null) {
processor.setCacheSize(num);
diff --git a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/WireTapReifier.java b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/WireTapReifier.java
index 08942ac00d5..ac8b651c097 100644
--- a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/WireTapReifier.java
+++ b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/WireTapReifier.java
@@ -84,8 +84,8 @@ public class WireTapReifier extends ToDynamicReifier<WireTapDefinition<?>> {
Endpoint endpoint = CamelContextHelper.resolveEndpoint(camelContext, uri, null);
LineNumberAware.trySetLineNumberAware(endpoint, definition);
sendProcessor = new SendProcessor(endpoint);
- sendProcessor.setVariableSend(definition.getVariableSend());
- sendProcessor.setVariableReceive(definition.getVariableReceive());
+ sendProcessor.setVariableSend(parseString(definition.getVariableSend()));
+ sendProcessor.setVariableReceive(parseString(definition.getVariableReceive()));
}
// create error handler we need to use for processing the wire tapped
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/PollEnrichVariableTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/PollEnrichVariableTest.java
new file mode 100644
index 00000000000..43109571c6e
--- /dev/null
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/PollEnrichVariableTest.java
@@ -0,0 +1,52 @@
+/*
+ * 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.processor;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+
+public class PollEnrichVariableTest extends ContextTestSupport {
+
+ @Test
+ public void testReceive() throws Exception {
+ template.sendBody("seda:foo", "Bye World");
+
+ getMockEndpoint("mock:after").expectedBodiesReceived("World");
+ getMockEndpoint("mock:after").expectedVariableReceived("bye", "Bye World");
+ getMockEndpoint("mock:result").expectedBodiesReceived("Bye World");
+ getMockEndpoint("mock:result").expectedVariableReceived("bye", "Bye World");
+
+ template.sendBody("direct:receive", "World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:receive")
+ .pollEnrich().constant("seda:foo").timeout(1000).variableReceive("bye")
+ .to("mock:after")
+ .setBody(simple("${variable:bye}"))
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPollEnricherMBean.java b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPollEnricherMBean.java
index f0486f10014..9e677be2500 100644
--- a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPollEnricherMBean.java
+++ b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedPollEnricherMBean.java
@@ -30,6 +30,9 @@ public interface ManagedPollEnricherMBean extends ManagedProcessorMBean, Managed
mask = true)
String getExpression();
+ @ManagedAttribute(description = "Variable to store the received message body (only body, not headers)")
+ String getVariableReceive();
+
@ManagedAttribute(description = "Timeout in millis when polling from the external service")
Long getTimeout();
diff --git a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPollEnricher.java b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPollEnricher.java
index 703e145c827..09d93a1b9d6 100644
--- a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPollEnricher.java
+++ b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPollEnricher.java
@@ -89,6 +89,11 @@ public class ManagedPollEnricher extends ManagedProcessor implements ManagedPoll
return uri;
}
+ @Override
+ public String getVariableReceive() {
+ return processor.getVariableReceive();
+ }
+
@Override
public Long getTimeout() {
return processor.getTimeout();
diff --git a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
index ccc33dcc6d1..6a4f5892c57 100644
--- a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
+++ b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/in/ModelParser.java
@@ -763,6 +763,7 @@ public class ModelParser extends BaseParser {
case "cacheSize": def.setCacheSize(val); break;
case "ignoreInvalidEndpoint": def.setIgnoreInvalidEndpoint(val); break;
case "timeout": def.setTimeout(val); break;
+ case "variableReceive": def.setVariableReceive(val); break;
default: return processorDefinitionAttributeHandler().accept(def, key, val);
}
return true;
diff --git a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
index b012cca083c..29a3ac921fe 100644
--- a/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
+++ b/core/camel-xml-io/src/generated/java/org/apache/camel/xml/out/ModelWriter.java
@@ -1762,6 +1762,7 @@ public class ModelWriter extends BaseWriter {
throws IOException {
startElement(name);
doWriteProcessorDefinitionAttributes(def);
+ doWriteAttribute("variableReceive", def.getVariableReceive());
doWriteAttribute("cacheSize", def.getCacheSize());
doWriteAttribute("aggregationStrategy", def.getAggregationStrategy());
doWriteAttribute("ignoreInvalidEndpoint", def.getIgnoreInvalidEndpoint());
diff --git a/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/ModelWriter.java b/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/ModelWriter.java
index cecac392932..974cebf1f05 100644
--- a/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/ModelWriter.java
+++ b/core/camel-yaml-io/src/generated/java/org/apache/camel/yaml/out/ModelWriter.java
@@ -1762,6 +1762,7 @@ public class ModelWriter extends BaseWriter {
throws IOException {
startElement(name);
doWriteProcessorDefinitionAttributes(def);
+ doWriteAttribute("variableReceive", def.getVariableReceive());
doWriteAttribute("cacheSize", def.getCacheSize());
doWriteAttribute("aggregationStrategy", def.getAggregationStrategy());
doWriteAttribute("ignoreInvalidEndpoint", def.getIgnoreInvalidEndpoint());
diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java b/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
index d0bf6091766..7ef2c0a270e 100644
--- a/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
+++ b/dsl/camel-yaml-dsl/camel-yaml-dsl-deserializers/src/generated/java/org/apache/camel/dsl/yaml/deserializers/ModelDeserializers.java
@@ -11283,7 +11283,8 @@ public final class ModelDeserializers extends YamlDeserializerSupport {
@YamlProperty(name = "id", type = "string", description = "Sets the id of this node", displayName = "Id"),
@YamlProperty(name = "ignoreInvalidEndpoint", type = "boolean", description = "Ignore the invalidate endpoint exception when try to create a producer with that endpoint", displayName = "Ignore Invalid Endpoint"),
@YamlProperty(name = "inheritErrorHandler", type = "boolean"),
- @YamlProperty(name = "timeout", type = "string", defaultValue = "-1", description = "Timeout in millis when polling from the external service. The timeout has influence about the poll enrich behavior. It basically operations in three different modes: negative value - Waits until a message is available and then returns it. Warning that this method could block indefinitely if no messages are available. 0 - Attempts to receive a message exchange immediately without waiti [...]
+ @YamlProperty(name = "timeout", type = "string", defaultValue = "-1", description = "Timeout in millis when polling from the external service. The timeout has influence about the poll enrich behavior. It basically operations in three different modes: negative value - Waits until a message is available and then returns it. Warning that this method could block indefinitely if no messages are available. 0 - Attempts to receive a message exchange immediately without waiti [...]
+ @YamlProperty(name = "variableReceive", type = "string", description = "To use a variable to store the received message body (only body, not headers). This is handy for easy access to the received message body via variables. Important: When using receive variable then the received body is stored only in this variable and not on the current org.apache.camel.Message .", displayName = "Variable Receive")
}
)
public static class PollEnrichDefinitionDeserializer extends YamlDeserializerBase<PollEnrichDefinition> {
@@ -11356,6 +11357,11 @@ public final class ModelDeserializers extends YamlDeserializerSupport {
target.setTimeout(val);
break;
}
+ case "variableReceive": {
+ String val = asText(node);
+ target.setVariableReceive(val);
+ break;
+ }
case "id": {
String val = asText(node);
target.setId(val);
diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
index 4f7c733fd1c..bafd77a84db 100644
--- a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
+++ b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/generated/resources/schema/camelYamlDsl.json
@@ -3759,6 +3759,11 @@
"description" : "Timeout in millis when polling from the external service. The timeout has influence about the poll enrich behavior. It basically operations in three different modes: negative value - Waits until a message is available and then returns it. Warning that this method could block indefinitely if no messages are available. 0 - Attempts to receive a message exchange immediately without waiting and returning null if a message exchange is not available yet. positive v [...]
"default" : "-1"
},
+ "variableReceive" : {
+ "type" : "string",
+ "title" : "Variable Receive",
+ "description" : "To use a variable to store the received message body (only body, not headers). This is handy for easy access to the received message body via variables. Important: When using receive variable then the received body is stored only in this variable and not on the current org.apache.camel.Message ."
+ },
"constant" : { },
"csimple" : { },
"datasonnet" : { },
diff --git a/dsl/camel-yaml-dsl/camel-yaml-dsl/src/test/groovy/org/apache/camel/dsl/yaml/PollEnrichVariableTest.groovy b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/test/groovy/org/apache/camel/dsl/yaml/PollEnrichVariableTest.groovy
new file mode 100644
index 00000000000..1552af9c881
--- /dev/null
+++ b/dsl/camel-yaml-dsl/camel-yaml-dsl/src/test/groovy/org/apache/camel/dsl/yaml/PollEnrichVariableTest.groovy
@@ -0,0 +1,64 @@
+/*
+ * 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.dsl.yaml
+
+import org.apache.camel.component.mock.MockEndpoint
+import org.apache.camel.dsl.yaml.support.YamlTestSupport
+
+class PollEnrichVariableTest extends YamlTestSupport {
+
+ def "pollEnrichVariable receive"() {
+ setup:
+ loadRoutes '''
+ - route:
+ from:
+ uri: direct:receive
+ steps:
+ - pollEnrich:
+ constant: "seda:foo"
+ timeout: "1000"
+ variableReceive: bye
+ - to:
+ uri: mock:after
+ - setBody:
+ simple: ${variable:bye}
+ - to:
+ uri: mock:result
+ '''
+
+ withMock('mock:after') {
+ expectedBodiesReceived 'World'
+ expectedVariableReceived("bye", "Bye World")
+ }
+ withMock('mock:result') {
+ expectedBodiesReceived 'Bye World'
+ expectedVariableReceived("bye", "Bye World")
+ }
+
+ when:
+ context.start()
+
+ withTemplate {
+ to('seda:foo').withBody('Bye World').send()
+ to('direct:receive').withBody('World').send()
+ }
+
+ then:
+ MockEndpoint.assertIsSatisfied(context)
+ }
+
+}