You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by nf...@apache.org on 2018/01/11 08:36:27 UTC
[camel] branch master updated (7a0ca73 -> fd1fbd3)
This is an automated email from the ASF dual-hosted git repository.
nferraro pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from 7a0ca73 Upgrade Box Java SDK to version 2.9.0
new 03550e5 CAMEL-11665: Saga EIP
new fd1fbd3 CAMEL-11665: fix configuration and documentation
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
apache-camel/src/main/descriptors/common-bin.xml | 2 +
bom/camel-bom/pom.xml | 10 +
camel-core/pom.xml | 1 +
camel-core/readme-eip.adoc | 5 +-
camel-core/readme.adoc | 5 +-
camel-core/src/main/docs/eips/saga-eip.adoc | 450 +++++++++++++++++++++
camel-core/src/main/docs/saga-component.adoc | 50 +++
.../src/main/java/org/apache/camel/Exchange.java | 4 +
.../apache/camel/component/saga/SagaComponent.java | 45 +++
.../apache/camel/component/saga/SagaEndpoint.java | 69 ++++
.../apache/camel/component/saga/SagaProducer.java | 78 ++++
.../camel/impl/saga/InMemorySagaCoordinator.java | 228 +++++++++++
.../camel/impl/saga/InMemorySagaService.java | 117 ++++++
.../org/apache/camel/model/PolicyDefinition.java | 5 +
.../apache/camel/model/ProcessorDefinition.java | 23 ++
.../camel/model/ProcessorDefinitionHelper.java | 2 +-
.../apache/camel/model/RouteDefinitionHelper.java | 25 ++
.../camel/model/SagaActionUriDefinition.java | 57 +++
.../org/apache/camel/model/SagaCompletionMode.java | 38 ++
.../org/apache/camel/model/SagaDefinition.java | 347 ++++++++++++++++
.../apache/camel/model/SagaOptionDefinition.java | 79 ++++
.../org/apache/camel/model/SagaPropagation.java | 54 +++
.../apache/camel/model/TransactedDefinition.java | 5 +
.../processor/saga/MandatorySagaProcessor.java | 52 +++
.../camel/processor/saga/NeverSagaProcessor.java | 56 +++
.../processor/saga/NotSupportedSagaProcessor.java | 56 +++
.../processor/saga/RequiredSagaProcessor.java | 69 ++++
.../processor/saga/RequiresNewSagaProcessor.java | 52 +++
.../apache/camel/processor/saga/SagaProcessor.java | 109 +++++
.../camel/processor/saga/SagaProcessorBuilder.java | 99 +++++
.../processor/saga/SupportsSagaProcessor.java | 54 +++
.../apache/camel/saga/CamelSagaCoordinator.java | 36 ++
.../org/apache/camel/saga/CamelSagaService.java | 35 ++
.../java/org/apache/camel/saga/CamelSagaStep.java | 65 +++
.../services/org/apache/camel/component/saga | 18 +
.../resources/org/apache/camel/model/jaxb.index | 1 +
.../camel/component/saga/SagaComponentTest.java | 132 ++++++
.../java/org/apache/camel/model/XmlParseTest.java | 8 +
.../apache/camel/processor/SagaFailuresTest.java | 131 ++++++
.../apache/camel/processor/SagaOptionsTest.java | 89 ++++
.../camel/processor/SagaPropagationTest.java | 197 +++++++++
.../java/org/apache/camel/processor/SagaTest.java | 200 +++++++++
.../apache/camel/processor/SagaTimeoutTest.java | 94 +++++
.../test/resources/org/apache/camel/model/saga.xml | 31 ++
components/camel-lra/pom.xml | 122 ++++++
components/camel-lra/src/main/docs/lra.adoc | 18 +
.../org/apache/camel/service/lra/LRAClient.java | 191 +++++++++
.../org/apache/camel/service/lra/LRAConstants.java | 43 ++
.../camel/service/lra/LRASagaCoordinator.java | 61 +++
.../apache/camel/service/lra/LRASagaRoutes.java | 89 ++++
.../apache/camel/service/lra/LRASagaService.java | 168 ++++++++
.../org/apache/camel/service/lra/LRASagaStep.java | 81 ++++
.../apache/camel/service/lra/LRAUrlBuilder.java | 139 +++++++
.../src/main/resources/META-INF/LICENSE.txt | 0
.../src/main/resources/META-INF/NOTICE.txt | 0
.../camel/service/lra/AbstractLRATestSupport.java | 105 +++++
.../org/apache/camel/service/lra/LRACreditIT.java | 196 +++++++++
.../apache/camel/service/lra/LRAFailuresIT.java | 104 +++++
.../org/apache/camel/service/lra/LRAManualIT.java | 95 +++++
.../org/apache/camel/service/lra/LRAOptionsIT.java | 87 ++++
.../org/apache/camel/service/lra/LRATimeoutIT.java | 94 +++++
.../camel-lra/src/test/resources/log4j2.properties | 29 ++
.../camel/spring/processor/SpringSagaTest.java | 61 +++
.../org/apache/camel/spring/processor/saga.xml | 57 +++
components/pom.xml | 1 +
components/readme.adoc | 16 +-
docs/user-manual/en/SUMMARY.md | 2 +
parent/pom.xml | 10 +
.../springboot/SagaComponentAutoConfiguration.java | 128 ++++++
.../springboot/SagaComponentConfiguration.java | 50 +++
.../src/main/resources/META-INF/spring.factories | 4 +-
.../components-starter/camel-lra-starter/pom.xml | 53 +++
.../springboot/LraServiceAutoConfiguration.java | 77 ++++
.../lra/springboot/LraServiceConfiguration.java | 94 +++++
.../src/main/resources/META-INF/LICENSE.txt | 0
.../src/main/resources/META-INF/NOTICE.txt | 0
.../src/main/resources/META-INF/spring.factories | 18 +
.../src/main/resources/META-INF/spring.provides | 17 +
platforms/spring-boot/components-starter/pom.xml | 1 +
.../camel-spring-boot-dependencies/pom.xml | 10 +
80 files changed, 5498 insertions(+), 6 deletions(-)
create mode 100644 camel-core/src/main/docs/eips/saga-eip.adoc
create mode 100644 camel-core/src/main/docs/saga-component.adoc
create mode 100644 camel-core/src/main/java/org/apache/camel/component/saga/SagaComponent.java
create mode 100644 camel-core/src/main/java/org/apache/camel/component/saga/SagaEndpoint.java
create mode 100644 camel-core/src/main/java/org/apache/camel/component/saga/SagaProducer.java
create mode 100644 camel-core/src/main/java/org/apache/camel/impl/saga/InMemorySagaCoordinator.java
create mode 100644 camel-core/src/main/java/org/apache/camel/impl/saga/InMemorySagaService.java
create mode 100644 camel-core/src/main/java/org/apache/camel/model/SagaActionUriDefinition.java
create mode 100644 camel-core/src/main/java/org/apache/camel/model/SagaCompletionMode.java
create mode 100644 camel-core/src/main/java/org/apache/camel/model/SagaDefinition.java
create mode 100644 camel-core/src/main/java/org/apache/camel/model/SagaOptionDefinition.java
create mode 100644 camel-core/src/main/java/org/apache/camel/model/SagaPropagation.java
create mode 100644 camel-core/src/main/java/org/apache/camel/processor/saga/MandatorySagaProcessor.java
create mode 100644 camel-core/src/main/java/org/apache/camel/processor/saga/NeverSagaProcessor.java
create mode 100644 camel-core/src/main/java/org/apache/camel/processor/saga/NotSupportedSagaProcessor.java
create mode 100644 camel-core/src/main/java/org/apache/camel/processor/saga/RequiredSagaProcessor.java
create mode 100644 camel-core/src/main/java/org/apache/camel/processor/saga/RequiresNewSagaProcessor.java
create mode 100644 camel-core/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java
create mode 100644 camel-core/src/main/java/org/apache/camel/processor/saga/SagaProcessorBuilder.java
create mode 100644 camel-core/src/main/java/org/apache/camel/processor/saga/SupportsSagaProcessor.java
create mode 100644 camel-core/src/main/java/org/apache/camel/saga/CamelSagaCoordinator.java
create mode 100644 camel-core/src/main/java/org/apache/camel/saga/CamelSagaService.java
create mode 100644 camel-core/src/main/java/org/apache/camel/saga/CamelSagaStep.java
create mode 100644 camel-core/src/main/resources/META-INF/services/org/apache/camel/component/saga
create mode 100644 camel-core/src/test/java/org/apache/camel/component/saga/SagaComponentTest.java
create mode 100644 camel-core/src/test/java/org/apache/camel/processor/SagaFailuresTest.java
create mode 100644 camel-core/src/test/java/org/apache/camel/processor/SagaOptionsTest.java
create mode 100644 camel-core/src/test/java/org/apache/camel/processor/SagaPropagationTest.java
create mode 100644 camel-core/src/test/java/org/apache/camel/processor/SagaTest.java
create mode 100644 camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java
create mode 100644 camel-core/src/test/resources/org/apache/camel/model/saga.xml
create mode 100644 components/camel-lra/pom.xml
create mode 100644 components/camel-lra/src/main/docs/lra.adoc
create mode 100644 components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAClient.java
create mode 100644 components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAConstants.java
create mode 100644 components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaCoordinator.java
create mode 100644 components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java
create mode 100644 components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java
create mode 100644 components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaStep.java
create mode 100644 components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAUrlBuilder.java
copy {tooling/maven/guice-maven-plugin => components/camel-lra}/src/main/resources/META-INF/LICENSE.txt (100%)
copy {tooling/maven/guice-maven-plugin => components/camel-lra}/src/main/resources/META-INF/NOTICE.txt (100%)
create mode 100644 components/camel-lra/src/test/java/org/apache/camel/service/lra/AbstractLRATestSupport.java
create mode 100644 components/camel-lra/src/test/java/org/apache/camel/service/lra/LRACreditIT.java
create mode 100644 components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAFailuresIT.java
create mode 100644 components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAManualIT.java
create mode 100644 components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAOptionsIT.java
create mode 100644 components/camel-lra/src/test/java/org/apache/camel/service/lra/LRATimeoutIT.java
create mode 100644 components/camel-lra/src/test/resources/log4j2.properties
create mode 100644 components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringSagaTest.java
create mode 100644 components/camel-spring/src/test/resources/org/apache/camel/spring/processor/saga.xml
create mode 100644 platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentAutoConfiguration.java
create mode 100644 platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentConfiguration.java
create mode 100644 platforms/spring-boot/components-starter/camel-lra-starter/pom.xml
create mode 100644 platforms/spring-boot/components-starter/camel-lra-starter/src/main/java/org/apache/camel/service/lra/springboot/LraServiceAutoConfiguration.java
create mode 100644 platforms/spring-boot/components-starter/camel-lra-starter/src/main/java/org/apache/camel/service/lra/springboot/LraServiceConfiguration.java
copy {tooling/maven/guice-maven-plugin => platforms/spring-boot/components-starter/camel-lra-starter}/src/main/resources/META-INF/LICENSE.txt (100%)
copy {tooling/maven/guice-maven-plugin => platforms/spring-boot/components-starter/camel-lra-starter}/src/main/resources/META-INF/NOTICE.txt (100%)
create mode 100644 platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/spring.factories
create mode 100644 platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/spring.provides
--
To stop receiving notification emails like this one, please contact
['"commits@camel.apache.org" <co...@camel.apache.org>'].
[camel] 01/02: CAMEL-11665: Saga EIP
Posted by nf...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nferraro pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 03550e5b440f3febd9dcdb44cd61f66864da67a7
Author: Nicola Ferraro <ni...@gmail.com>
AuthorDate: Wed Aug 9 16:50:19 2017 +0200
CAMEL-11665: Saga EIP
---
apache-camel/src/main/descriptors/common-bin.xml | 2 +
bom/camel-bom/pom.xml | 10 +
camel-core/pom.xml | 1 +
camel-core/readme-eip.adoc | 5 +-
camel-core/readme.adoc | 5 +-
camel-core/src/main/docs/eips/saga-eip.adoc | 450 +++++++++++++++++++++
camel-core/src/main/docs/saga-component.adoc | 52 +++
.../src/main/java/org/apache/camel/Exchange.java | 3 +
.../apache/camel/component/saga/SagaComponent.java | 45 +++
.../apache/camel/component/saga/SagaEndpoint.java | 69 ++++
.../apache/camel/component/saga/SagaProducer.java | 78 ++++
.../camel/impl/saga/InMemorySagaCoordinator.java | 228 +++++++++++
.../camel/impl/saga/InMemorySagaService.java | 117 ++++++
.../org/apache/camel/model/PolicyDefinition.java | 5 +
.../apache/camel/model/ProcessorDefinition.java | 23 ++
.../camel/model/ProcessorDefinitionHelper.java | 2 +-
.../apache/camel/model/RouteDefinitionHelper.java | 25 ++
.../camel/model/SagaActionUriDefinition.java | 57 +++
.../org/apache/camel/model/SagaCompletionMode.java | 38 ++
.../org/apache/camel/model/SagaDefinition.java | 347 ++++++++++++++++
.../apache/camel/model/SagaOptionDefinition.java | 79 ++++
.../org/apache/camel/model/SagaPropagation.java | 54 +++
.../apache/camel/model/TransactedDefinition.java | 5 +
.../processor/saga/MandatorySagaProcessor.java | 52 +++
.../camel/processor/saga/NeverSagaProcessor.java | 56 +++
.../processor/saga/NotSupportedSagaProcessor.java | 56 +++
.../processor/saga/RequiredSagaProcessor.java | 69 ++++
.../processor/saga/RequiresNewSagaProcessor.java | 52 +++
.../apache/camel/processor/saga/SagaProcessor.java | 109 +++++
.../camel/processor/saga/SagaProcessorBuilder.java | 99 +++++
.../processor/saga/SupportsSagaProcessor.java | 54 +++
.../apache/camel/saga/CamelSagaCoordinator.java | 36 ++
.../org/apache/camel/saga/CamelSagaService.java | 35 ++
.../java/org/apache/camel/saga/CamelSagaStep.java | 65 +++
.../services/org/apache/camel/component/saga | 18 +
.../resources/org/apache/camel/model/jaxb.index | 1 +
.../camel/component/saga/SagaComponentTest.java | 132 ++++++
.../java/org/apache/camel/model/XmlParseTest.java | 8 +
.../apache/camel/processor/SagaFailuresTest.java | 131 ++++++
.../apache/camel/processor/SagaOptionsTest.java | 89 ++++
.../camel/processor/SagaPropagationTest.java | 197 +++++++++
.../java/org/apache/camel/processor/SagaTest.java | 200 +++++++++
.../apache/camel/processor/SagaTimeoutTest.java | 94 +++++
.../test/resources/org/apache/camel/model/saga.xml | 31 ++
components/camel-lra/pom.xml | 122 ++++++
components/camel-lra/src/main/docs/lra.adoc | 18 +
.../org/apache/camel/service/lra/LRAClient.java | 191 +++++++++
.../org/apache/camel/service/lra/LRAConstants.java | 43 ++
.../camel/service/lra/LRASagaCoordinator.java | 61 +++
.../apache/camel/service/lra/LRASagaRoutes.java | 89 ++++
.../apache/camel/service/lra/LRASagaService.java | 168 ++++++++
.../org/apache/camel/service/lra/LRASagaStep.java | 81 ++++
.../apache/camel/service/lra/LRAUrlBuilder.java | 139 +++++++
.../src/main/resources/META-INF/LICENSE.txt | 203 ++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 +
.../camel/service/lra/AbstractLRATestSupport.java | 105 +++++
.../org/apache/camel/service/lra/LRACreditIT.java | 196 +++++++++
.../apache/camel/service/lra/LRAFailuresIT.java | 104 +++++
.../org/apache/camel/service/lra/LRAManualIT.java | 95 +++++
.../org/apache/camel/service/lra/LRAOptionsIT.java | 87 ++++
.../org/apache/camel/service/lra/LRATimeoutIT.java | 94 +++++
.../camel-lra/src/test/resources/log4j2.properties | 29 ++
.../camel/spring/processor/SpringSagaTest.java | 61 +++
.../org/apache/camel/spring/processor/saga.xml | 57 +++
components/pom.xml | 1 +
components/readme.adoc | 9 +-
docs/user-manual/en/SUMMARY.md | 2 +
parent/pom.xml | 10 +
.../springboot/SagaComponentAutoConfiguration.java | 128 ++++++
.../springboot/SagaComponentConfiguration.java | 49 +++
.../src/main/resources/META-INF/spring.factories | 4 +-
.../components-starter/camel-lra-starter/pom.xml | 53 +++
.../springboot/LraServiceAutoConfiguration.java | 77 ++++
.../lra/springboot/LraServiceConfiguration.java | 94 +++++
.../src/main/resources/META-INF/LICENSE.txt | 203 ++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 +
.../src/main/resources/META-INF/spring.factories | 18 +
.../src/main/resources/META-INF/spring.provides | 17 +
platforms/spring-boot/components-starter/pom.xml | 1 +
.../camel-spring-boot-dependencies/pom.xml | 10 +
80 files changed, 5919 insertions(+), 6 deletions(-)
diff --git a/apache-camel/src/main/descriptors/common-bin.xml b/apache-camel/src/main/descriptors/common-bin.xml
index 2eb3d3c..7a80a11 100644
--- a/apache-camel/src/main/descriptors/common-bin.xml
+++ b/apache-camel/src/main/descriptors/common-bin.xml
@@ -172,6 +172,7 @@
<include>org.apache.camel:camel-leveldb</include>
<include>org.apache.camel:camel-linkedin-api</include>
<include>org.apache.camel:camel-linkedin</include>
+ <include>org.apache.camel:camel-lra</include>
<include>org.apache.camel:camel-lucene</include>
<include>org.apache.camel:camel-lumberjack</include>
<include>org.apache.camel:camel-lzf</include>
@@ -478,6 +479,7 @@
<include>org.apache.camel:camel-ldif-starter</include>
<include>org.apache.camel:camel-leveldb-starter</include>
<include>org.apache.camel:camel-linkedin-starter</include>
+ <include>org.apache.camel:camel-lra-starter</include>
<include>org.apache.camel:camel-lucene-starter</include>
<include>org.apache.camel:camel-lumberjack-starter</include>
<include>org.apache.camel:camel-lzf-starter</include>
diff --git a/bom/camel-bom/pom.xml b/bom/camel-bom/pom.xml
index a0b5452..bc85425 100644
--- a/bom/camel-bom/pom.xml
+++ b/bom/camel-bom/pom.xml
@@ -1555,6 +1555,16 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-lra</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-lra-starter</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-lucene</artifactId>
<version>${project.version}</version>
</dependency>
diff --git a/camel-core/pom.xml b/camel-core/pom.xml
index 3c1bf09..c92265c 100644
--- a/camel-core/pom.xml
+++ b/camel-core/pom.xml
@@ -77,6 +77,7 @@
org.apache.camel.spi.ComponentResolver;component=rest,
org.apache.camel.spi.ComponentResolver;component=rest-api,
org.apache.camel.spi.ComponentResolver;component=scheduler,
+ org.apache.camel.spi.ComponentResolver;component=saga,
org.apache.camel.spi.ComponentResolver;component=seda,
org.apache.camel.spi.ComponentResolver;component=stub,
org.apache.camel.spi.ComponentResolver;component=test,
diff --git a/camel-core/readme-eip.adoc b/camel-core/readme-eip.adoc
index 40290bf..c7474f3 100644
--- a/camel-core/readme-eip.adoc
+++ b/camel-core/readme-eip.adoc
@@ -4,7 +4,7 @@ Enterprise Integration Patterns
Camel supports most of the link:http://www.eaipatterns.com/toc.html[Enterprise Integration Patterns] from the excellent book by link:http://www.amazon.com/exec/obidos/search-handle-url/105-9796798-8100401?%5Fencoding=UTF8&search-type=ss&index=books&field-author=Gregor%20Hohpe[Gregor Hohpe] and link:http://www.amazon.com/exec/obidos/search-handle-url/105-9796798-8100401?%5Fencoding=UTF8&search-type=ss&index=books&field-author=Bobby%20Woolf[Bobby Woolf].
// eips: START
-Number of EIPs: 62 (2 deprecated)
+Number of EIPs: 63 (2 deprecated)
[width="100%",cols="4,6",options="header"]
|===
@@ -124,6 +124,9 @@ Number of EIPs: 62 (2 deprecated)
| link:src/main/docs/eips/routingSlip-eip.adoc[Routing Slip] +
`<routingSlip>` | Routes a message through a series of steps that are pre-determined (the slip)
+| link:src/main/docs/eips/saga-eip.adoc[Saga] +
+`<saga>` | Enables sagas on the route
+
| link:src/main/docs/eips/sample-eip.adoc[Sample] +
`<sample>` | Extract a sample of the messages passing through a route
diff --git a/camel-core/readme.adoc b/camel-core/readme.adoc
index 83d34b8..50eee0e 100644
--- a/camel-core/readme.adoc
+++ b/camel-core/readme.adoc
@@ -6,7 +6,7 @@ Components
// components: START
-Number of Components: 25 in 1 JAR artifacts (1 deprecated)
+Number of Components: 26 in 1 JAR artifacts (1 deprecated)
[width="100%",cols="4,1,5",options="header"]
|===
@@ -63,6 +63,9 @@ Number of Components: 25 in 1 JAR artifacts (1 deprecated)
| link:src/main/docs/rest-api-component.adoc[REST API] (camel-core) +
`rest-api:path/contextIdPattern` | 2.16 | The rest-api component is used for providing Swagger API of the REST services which has been defined using the rest-dsl in Camel.
+| link:src/main/docs/saga-component.adoc[Saga] (camel-core) +
+`saga:action` | 2.21 | The default saga endpoint.
+
| link:src/main/docs/scheduler-component.adoc[Scheduler] (camel-core) +
`scheduler:name` | 2.15 | The scheduler component is used for generating message exchanges when a scheduler fires.
diff --git a/camel-core/src/main/docs/eips/saga-eip.adoc b/camel-core/src/main/docs/eips/saga-eip.adoc
new file mode 100644
index 0000000..935baed
--- /dev/null
+++ b/camel-core/src/main/docs/eips/saga-eip.adoc
@@ -0,0 +1,450 @@
+== Saga EIP
+
+*Available as of Camel 2.21*
+
+The Saga EIP provides a way to define a series of related actions in a Camel route that should be either completed successfully (*all of them*) or not-executed/compensated.
+Sagas implementations are able to coordinate *distributed services communicating using any transport* towards a globally *consistent outcome*.
+
+Although their main purpose is similar, Sagas are different from classical ACID distributed (XA) transactions because the status of the different participating services is guaranteed to be consistent
+only at the end of the Saga and not in any intermediate step (lack of isolation).
+
+Conversely, Sagas are suitable for many use cases where usage of distributed transactions is discouraged.
+For example, services participating in a Saga are allowed to use any kind of datastore: classical databases or even NoSQL non-transactional datastores.
+Sagas are also suitable for being used in stateless cloud services as they do not require a transaction log
+to be stored alongside the service.
+
+Differently from transactions, Sagas are also not required to be completed in a small amount of time, because they don't use database-level locks. They can live for a longer timespan: from few seconds to several days.
+The Saga EIP implementation based on the Microprofile sandbox spec (see camel-lra) is indeed called LRA that stands for "Long Running Action".
+It also supports coordination of external *heterogeneous services*, written with any language/technology and also running outside a JVM.
+
+Sagas don't use locks on data, instead they define the concept of "Compensating Action" that is an action that should be executed when the standard flow encounters an error,
+with the purpose of restoring the status that was present before the flow execution.
+Compensating actions can be declared in Camel routes using the Java or XML DSL and will be invoked by Camel only when needed (if the saga is cancelled due to an error).
+
+// eip options: START
+The Saga EIP supports 6 options which are listed below:
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *propagation* | Set the Saga propagation mode (REQUIRED REQUIRES_NEW MANDATORY SUPPORTS NOT_SUPPORTED NEVER). | REQUIRED | SagaPropagation
+| *completionMode* | Determine how the saga should be considered complete. When set to AUTO the saga is completed when the exchange that initiates the saga is processed successfully or compensated when it completes exceptionally. When set to MANUAL the user must complete or compensate the saga using the saga:complete or saga:compensate endpoints. | AUTO | SagaCompletionMode
+| *timeoutInMilliseconds* | Set the maximum amount of time for the Saga. After the timeout is expired the saga will be compensated automatically (unless a different decision has been taken in the meantime). | | Long
+| *compensation* | The compensation endpoint URI that must be called to compensate all changes done in the route. The route corresponding to the compensation URI must perform compensation and complete without error. If errors occur during compensation the saga service may call again the compensation URI to retry. | | SagaActionUri Definition
+| *completion* | The completion endpoint URI that will be called when the Saga is completed successfully. The route corresponding to the completion URI must perform completion tasks and terminate without error. If errors occur during completion the saga service may call again the completion URI to retry. | | SagaActionUri Definition
+| *option* | Allows to save properties of the current exchange in order to re-use them in a compensation/completion callback route. Options are usually helpful e.g. to store and retrieve identifiers of objects that should be deleted in compensating actions. Option values will be transformed into input headers of the compensation/completion exchange. | | List
+|===
+// eip options: END
+
+=== Exchange properties
+The following properties are set on each Exchange that is participating to a Saga (normal actions, compensating actions and completions):
+
+[width="100%",cols="4m,2m,5",options="header"]
+|=======================================================================
+| Property | Type | Description
+| `Long-Running-Action` | `String` | A globally unique identifier for the Saga that can be propagated to remote systems using transport-level headers (e.g. HTTP).
+|=======================================================================
+
+=== Saga Service Configuration
+The Saga EIP requires that a service implementing the interface `org.apache.camel.saga.CamelSagaService` is added to the Camel context.
+
+Camel currently supports the following Saga Services:
+
+* *InMemorySagaService*: it is a *basic* implementation of the Saga EIP that does not support advanced features (no remote context propagation, no consistency guarantee in case of application failure).
+* *LRASagaService*: it is a *fully-fledged* implementation of the Saga EIP based on Microprofile sandbox LRA specification that supports remote context propagation and provides consistency guarantees in case of application failure.
+
+==== Using the In-Memory Saga Service
+
+The in-memory Saga service is not recommended for production environments as it does not support persistence of the Saga status (it is kept only in-memory),
+so it cannot guarantee consistency of Sagas in case of application failure (e.g. JVM crash).
+
+Also, when using a in-memory Saga service, Saga contexts cannot be propagated to remote services using transport-level headers (it can be done with other implementations).
+
+Users that want to use the in-memory saga service should add the following code to customize the Camel context.
+
+[source,java]
+----
+context.addService(new org.apache.camel.impl.saga.InMemorySagaService());
+----
+
+The service belongs to the `camel-core` module.
+
+==== Using the LRA Saga Service
+
+The LRA Saga Service is an implementation based on the Microprofile sandbox LRA specification.
+It leverages a *external Saga coordinator* to control the execution of the various steps of the Saga.
+The proposed reference implementation for the LRA specification is the http://jbossts.blogspot.it/2017/12/narayana-lra-implementation-of-saga.html[Narayana LRA Coordinator].
+Users can follows instructions present on the Narayana website to *startup a remote instance of the coordinator*.
+
+The URL of the LRA coordinator is a required parameter of the Camel LRA service. The Camel application and the LRA service communicate using the HTTP protocol.
+
+In order to use the LRA Saga service, maven users will need to add the following dependency to their pom.xml
+
+[source,xml]
+----
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-lra</artifactId>
+ <!-- use the same version as your Camel core version -->
+ <version>x.y.z</version>
+</dependency>
+----
+
+A Camel REST context is also required to be present for the LRA implementation to work. You may add `camel-undertow` for example.
+
+[source,xml]
+----
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-undertow</artifactId>
+ <!-- use the same version as your Camel core version -->
+ <version>x.y.z</version>
+</dependency>
+----
+
+[NOTE]
+====
+The LRA implementation of the Saga EIP will add some web endpoints under the "/lra-participant" path.
+Those endpoints will be used by the LRA coordinator for calling back the application.
+====
+
+[source,java]
+----
+// Configure the LRA saga service
+org.apache.camel.service.lra.LRASagaService sagaService = new org.apache.camel.service.lra.LRASagaService();
+sagaService.setCoordinatorUrl("http://lra-service-host");
+sagaService.setLocalParticipantUrl("http://my-host-as-seen-by-lra-service:8080/context-path");
+
+// Add it to the Camel context
+context.addService(sagaService);
+----
+
+===== Using the LRA Saga Service in Spring-Boot
+
+Spring-Boot users can use a simplified configuration model for the LRA Saga Service. Maven users can
+include the *camel-lra-starter* module in their project:
+
+[source,xml]
+----
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-lra-starter</artifactId>
+ <!-- use the same version as your Camel core version -->
+ <version>x.y.z</version>
+</dependency>
+
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-undertow-starter</artifactId>
+ <!-- use the same version as your Camel core version -->
+ <version>x.y.z</version>
+</dependency>
+----
+
+Configuration can be done in the Spring-Boot `application.yaml` file:
+
+.application.yaml
+[source,yaml]
+----
+camel:
+ service:
+ lra:
+ enabled: true
+ coordinator-url: http://lra-service-host
+ local-participant-url: http://my-host-as-seen-by-lra-service:8080/context-path
+----
+
+Once done, the Saga EIP can be directly used inside Camel routes and it will use the LRA Saga Service under the hood.
+
+=== Examples
+
+Suppose you want to place a new order and you have two distinct services in your system: one managing the orders and one managing the credit.
+Logically you can place a order if you have enough credit for it.
+
+With the Saga EIP you can model the _direct:buy_ route as a Saga composed of two distinct actions, one to create the order and one to take the credit.
+*Both actions must be executed, or none of them*: a order placed without credit can be considered a inconsistent outcome (as well as a payment without an order).
+
+[source,java]
+----
+from("direct:buy")
+ .saga()
+ .to("direct:newOrder")
+ .to("direct:reserveCredit");
+----
+
+*That's it*. The buy action will not change for the rest of the examples. We'll just see different options that can be used to model the "New Order" and "Reserve Credit" actions in the following.
+
+[NOTE]
+We have used a _direct_ endpoint to model the two actions since this example can be used with both implementations of the Saga service,
+but we could have used *http* or other kinds of endpoint with the LRA Saga service.
+
+Both services called by the _direct:buy_ route can *participate to the Saga* and declare their compensating actions.
+
+[source,java]
+----
+from("direct:newOrder")
+ .saga()
+ .propagation(SagaPropagation.MANDATORY)
+ .compensation("direct:cancelOrder")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(orderManagerService, "newOrder")
+ .log("Order ${body} created");
+----
+
+Here the propagation mode is set to _MANDATORY_ meaning that any exchange flowing in this route must be already part of a saga
+(and it is the case in this example, since the saga is created in the _direct:buy_ route).
+
+The _direct:newOrder_ route declares a compensating action that is called _direct:cancelOrder_, responsible for undoing the order in case the saga is cancelled.
+
+Each exchange always contains a `Exchange.SAGA_LONG_RUNNING_ACTION` header that here is used as id of the order.
+This is done in order to identify the order to delete in the corresponding compensating action, but it is not a requirement (options can be used as alternative solution).
+
+The compensating action of _direct:newOrder_ is _direct:cancelOrder_ and it's shown below:
+
+[source,java]
+----
+from("direct:cancelOrder")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(orderManagerService, "cancelOrder")
+ .log("Order ${body} cancelled");
+----
+
+It is called automatically by the Saga EIP implementation when the order should be cancelled.
+
+It should not terminate with error. In case an error is thrown in the _direct:cancelOrder_ route, the EIP implementation should
+periodically retry to execute the compensating action up to a certain limit.
+This means that *any compensating action must be idempotent*, so it should take into account that it may be triggered multiple times and should not fail in any case.
+
+If compensation cannot be done after all retries, a manual intervention process should be triggered by the Saga implementation.
+
+[NOTE]
+====
+It may happen that due to a delay in the execution of the _direct:newOrder_ route the Saga is cancelled by another party in the meantime (due to an error in a parallel route or a timeout at Saga level).
+
+So, when the compensating action _direct:cancelOrder_ is called, it may not find the Order record that should be cancelled.
+It is important, in order to guarantee full global consistency, that *any main action and its corresponding compensating action are commutative*,
+i.e. if compensation occurs before the main action it shoud have the same effect.
+
+Another possible approach, when using a commutative behavior is not possible,
+is to consistently fail in the compensating action until data produced by the main action is found (or the maximum number of retries is exhausted):
+this approach may work in many contexts, but it's *heuristic*.
+====
+
+The credit service may be implemented almost in the same way as the order service.
+
+[source,java]
+----
+// action
+from("direct:reserveCredit")
+ .saga()
+ .propagation(SagaPropagation.MANDATORY)
+ .compensation("direct:refundCredit")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(creditService, "reserveCredit")
+ .log("Credit ${header.amount} reserved in action ${body}");
+
+// compensation
+from("direct:refundCredit")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(creditService, "refundCredit")
+ .log("Credit for action ${body} refunded");
+----
+
+Here the compensating action for a credit reservation is a refund.
+
+This completes the example. It can be run with both implementations of the Saga EIP, as it does not involve remote endpoints.
+
+Further options will be shown next.
+
+==== Handling Completion Events
+It is often required to do some processing when the Saga is completed. Compensation endpoints are invoked when something wrong happens and the Saga is cancelled.
+Equivalently, *completion endpoints* can be invoked to do further processing when the Saga is completed successfully.
+
+For example, in the order service above, we may need to know when the order is completed (and the credit reserved) to actually start preparing the order.
+We will not want to start to prepare the order if the payment is not done (unlike most modern CPUs that give you access to reserved memory before ensuring that you have rights to read it).
+
+This can be done easily with a modified version of the _direct:newOrder_ endpoint:
+
+
+[source,java]
+----
+from("direct:newOrder")
+ .saga()
+ .propagation(SagaPropagation.MANDATORY)
+ .compensation("direct:cancelOrder")
+ .completion("direct:completeOrder") // completion endpoint
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(orderManagerService, "newOrder")
+ .log("Order ${body} created");
+
+// direct:cancelOrder is the same as in the previous example
+
+// called on successful completion
+from("direct:completeOrder")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(orderManagerService, "findExternalId")
+ .to("jms:prepareOrder")
+ .log("Order ${body} sent for preparation");
+----
+
+When the Saga is completed, the order is sent to a JMS queue for preparation.
+
+Like compensating actions, also completion actions may be called multiple times by the Saga coordinator (especially in case of errors, like network errors).
+In this example, the service listening to the _prepareOrder_ JMS queue should be prepared to hold possible duplicates (see the Idempotent Consumer EIP for examples on how to handle duplicates).
+
+==== Using Custom Identifiers and Options
+The example shown so far use the `Exchange.SAGA_LONG_RUNNING_ACTION` as identifier for the resources (order and credit).
+This is not always a desired approach, as it may pollute the business logic and the data model.
+
+An alternative approach is to use Saga options to "register" custom identifiers.
+For example, the credit service may be refactored as follows:
+
+[source,java]
+----
+// action
+from("direct:reserveCredit")
+ .bean(idService, "generateCustomId") // generate a custom Id and set it in the body
+ .to("direct:creditReservation")
+
+// delegate action
+from("direct:creditReservation")
+ .saga()
+ .propagation(SagaPropagation.SUPPORTS)
+ .option("CreditId", body()) // mark the current body as needed in the compensating action
+ .compensation("direct:creditRefund")
+ .bean(creditService, "reserveCredit")
+ .log("Credit ${header.amount} reserved. Custom Id used is ${body}");
+
+// called only if the saga is cancelled
+from("direct:creditRefund")
+ .transform(header("CreditId")) // retrieve the CreditId option from headers
+ .bean(creditService, "refundCredit")
+ .log("Credit for Custom Id ${body} refunded");
+----
+
+*Note how the previous listing is not using the `Exchange.SAGA_LONG_RUNNING_ACTION` header at all.*
+
+Since the _direct:creditReservation_ endpoint can be now called also from outside a Saga, the propagation mode can be set to *SUPPORTS*.
+
+*Multiple options* can be declared in a Saga route.
+
+==== Setting Timeouts
+Sagas are long running actions, but this does not mean that they should not have a bounded timeframe to execute.
+*Setting timeouts on Sagas is always a good practice* as it guarantees that a Saga does not remain stuck forever in the case of machine failure.
+
+NOTE: the Saga EIP implementation may have a default timeout set on all Sagas that don't specify it explicitly
+
+When the timeout expires, the Saga EIP will decide to *cancel the Saga* (and compensate all participants), unless a different decision has been taken before.
+
+Timeouts can be set on Saga participants as follows:
+
+[source,java]
+----
+from("direct:newOrder")
+ .saga()
+ .timeout(1, TimeUnit.MINUTES) // newOrder requires that the saga is completed within 1 minute
+ .propagation(SagaPropagation.MANDATORY)
+ .compensation("direct:cancelOrder")
+ .completion("direct:completeOrder")
+ // ...
+ .log("Order ${body} created");
+----
+
+All participants (e.g. credit service, order service) can set their own timeout. The minimum value of those timeouts is taken as timeout for the saga when they are composed together.
+
+A timeout can also be specified at saga level as follows:
+
+[source,java]
+----
+from("direct:buy")
+ .saga()
+ .timeout(5, TimeUnit.MINUTES) // timeout at saga level
+ .to("direct:newOrder")
+ .to("direct:reserveCredit");
+----
+
+==== Choosing Propagation
+In the examples above, we have used the _MANDATORY_ and _SUPPORTS_ propagation modes, but also the _REQUIRED_ propagation mode,
+that is the default propagation used when nothing else is specified.
+
+These propagation modes map 1:1 the equivalent modes used in transactional contexts. Here's a summary of their meaning:
+
+[width="100%",cols="2m,8",options="header"]
+|=======================================================================
+| Propagation | Description
+| `REQUIRED` | Join the existing saga or create a new one if it does not exist.
+| `REQUIRES_NEW` | Always create a new saga. Suspend the old saga and resume it when the new one terminates.
+| `MANDATORY` | A saga must be already present. The existing saga is joined.
+| `SUPPORTS` | If a saga already exists, then join it.
+| `NOT_SUPPORTED` | If a saga already exists, it is suspended and resumed when the current block completes.
+| `NEVER` | The current block must never be invoked within a saga.
+|=======================================================================
+
+==== Using Manual Completion (Advanced)
+When a Saga cannot be all executed in a synchronous way, but it requires e.g. communication with external services using asynchronous communication channels,
+the completion mode cannot be set to _AUTO_ (default), because the saga is not completed when the exchange that creates it is done.
+
+This is often the case for Sagas that have long execution times (hours, days). In these cases, the _MANUAL_ completion mode should be used.
+
+
+[source,java]
+----
+from("direct:mysaga")
+ .saga()
+ .completionMode(SagaCompletionMode.MANUAL)
+ .completion("direct:finalize")
+ .timeout(2, TimeUnit.HOURS)
+ .to("seda:newOrder")
+ .to("seda:reserveCredit");
+
+// Put here asynchronous processing for seda:newOrder and seda:reserveCredit
+// They will send asynchronous callbacks to seda:operationCompleted
+
+from("seda:operationCompleted") // an asynchronous callback
+ .saga()
+ .propagation(SagaPropagation.MANDATORY)
+ .bean(controlService, "actionExecuted")
+ .choice()
+ .when(body().isEqualTo("ok"))
+ .to("saga:complete") // complete the current saga manually (saga component)
+ .end()
+
+// You can put here the direct:finalize endpoint to execute final actions
+----
+
+Setting the completion mode to _MANUAL_ means that the saga is not completed when the exchange is processed in the route _direct:mysaga_ but
+it will last longer (max duration is set to 2 hours).
+
+When both asynchronous actions are completed the saga is completed. The call to complete is done using the Camel Saga Component's _saga:complete_ endpoint.
+There's is a similar endpoint for manually compensating the Saga (_saga:compensate_).
+
+Apparently the addition of the saga markers do not add much value to the flow: it works also if you remove all Saga EIP configuration.
+But Sagas add a lot of value, since they guarantee that even in the presence of unexpected issues (servers crashing, messages are lost)
+there will always be a consistent outcome: order placed and credit reserved, or none of them changed.
+In particular, if the Saga is not completed within 2 hours, the compensation mechanism will take care of fixing the status.
+
+=== XML Configuration
+
+Saga features are also available for users that want to use the XML configuration.
+
+The following snipped shows an example:
+
+[source,xml]
+----
+<route>
+ <from uri="direct:start"/>
+ <saga>
+ <compensation uri="direct:compensation" />
+ <completion uri="direct:completion" />
+ <option optionName="myOptionKey">
+ <constant>myOptionValue</constant>
+ </option>
+ <option optionName="myOptionKey2">
+ <constant>myOptionValue2</constant>
+ </option>
+ </saga>
+ <to uri="direct:action1" />
+ <to uri="direct:action2" />
+</route>
+----
diff --git a/camel-core/src/main/docs/saga-component.adoc b/camel-core/src/main/docs/saga-component.adoc
new file mode 100644
index 0000000..2610f4b
--- /dev/null
+++ b/camel-core/src/main/docs/saga-component.adoc
@@ -0,0 +1,52 @@
+== Saga Component
+
+*Available as of Camel version 2.21*
+
+The *saga* component provides a bridge to execute custom actions within a route using the Saga EIP.
+
+The component should be used for advanced tasks, such as deciding to complete or compensate a Saga
+with completionMode set to *MANUAL*.
+
+Refer to the Saga EIP documentation for help on using sagas in common scenarios.
+
+=== URI format
+
+[source]
+----
+saga:action
+----
+
+Where *`action`* can be ...
+
+=== Options
+
+// component options: START
+The Saga component has no options.
+// component options: END
+
+
+// endpoint options: START
+The Saga endpoint is configured using URI syntax:
+
+----
+saga:action
+----
+
+with the following path and query parameters:
+
+==== Path Parameters (1 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *action* | *Required* Action to execute (complete or compensate) | | SagaEndpointAction
+|===
+
+==== Query Parameters (1 parameters):
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported). | false | boolean
+|===
+// endpoint options: END
diff --git a/camel-core/src/main/java/org/apache/camel/Exchange.java b/camel-core/src/main/java/org/apache/camel/Exchange.java
index b1d232d..f3e3774 100644
--- a/camel-core/src/main/java/org/apache/camel/Exchange.java
+++ b/camel-core/src/main/java/org/apache/camel/Exchange.java
@@ -170,6 +170,9 @@ public interface Exchange {
String LOOP_INDEX = "CamelLoopIndex";
String LOOP_SIZE = "CamelLoopSize";
+ // Long running action (saga)
+ String SAGA_LONG_RUNNING_ACTION = "Long-Running-Action";
+
String MAXIMUM_CACHE_POOL_SIZE = "CamelMaximumCachePoolSize";
String MAXIMUM_ENDPOINT_CACHE_SIZE = "CamelMaximumEndpointCacheSize";
String MAXIMUM_SIMPLE_CACHE_SIZE = "CamelMaximumSimpleCacheSize";
diff --git a/camel-core/src/main/java/org/apache/camel/component/saga/SagaComponent.java b/camel-core/src/main/java/org/apache/camel/component/saga/SagaComponent.java
new file mode 100644
index 0000000..55414ed
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/component/saga/SagaComponent.java
@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.saga;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.DefaultComponent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The saga component for interacting with the saga processing framework.
+ */
+public class SagaComponent extends DefaultComponent {
+
+ protected final Logger log = LoggerFactory.getLogger(getClass());
+
+ public SagaComponent() {
+ }
+
+ public SagaComponent(CamelContext camelContext) {
+ super(camelContext);
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ return new SagaEndpoint(uri, this, remaining);
+ }
+}
diff --git a/camel-core/src/main/java/org/apache/camel/component/saga/SagaEndpoint.java b/camel-core/src/main/java/org/apache/camel/component/saga/SagaEndpoint.java
new file mode 100644
index 0000000..96cd167
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/component/saga/SagaEndpoint.java
@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.saga;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriPath;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * The default saga endpoint.
+ */
+@UriEndpoint(firstVersion = "2.21.0", scheme = "saga", title = "Saga", syntax = "saga:action", producerOnly = true, label = "core,endpoint")
+public class SagaEndpoint extends DefaultEndpoint {
+
+ public enum SagaEndpointAction {
+ COMPLETE,
+ COMPENSATE
+ }
+
+ @UriPath(description = "Action to execute (complete or compensate)")
+ @Metadata(required = "true")
+ private final SagaEndpointAction action;
+
+ public SagaEndpoint(String endpointUri, SagaComponent component, String action) {
+ super(endpointUri, component);
+ this.action = SagaEndpointAction.valueOf(ObjectHelper.notNull(action, "action").toUpperCase());
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ if (SagaEndpointAction.COMPLETE.equals(this.action)) {
+ return new SagaProducer(this, true);
+ } else if (SagaEndpointAction.COMPENSATE.equals(this.action)) {
+ return new SagaProducer(this, false);
+ } else {
+ throw new IllegalStateException("Unsupported action '" + this.action + "' in saga endpoint");
+ }
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ throw new UnsupportedOperationException("Consumer not allowed for saga endpoint");
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return false;
+ }
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/component/saga/SagaProducer.java b/camel-core/src/main/java/org/apache/camel/component/saga/SagaProducer.java
new file mode 100644
index 0000000..830e96e
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/component/saga/SagaProducer.java
@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.saga;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultAsyncProducer;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.util.CamelContextHelper;
+
+/**
+ * A producer that finalizes the current saga.
+ */
+public class SagaProducer extends DefaultAsyncProducer {
+
+ private final boolean success;
+
+ private CamelSagaService camelSagaService;
+
+ public SagaProducer(SagaEndpoint endpoint, boolean success) {
+ super(endpoint);
+ this.success = success;
+
+ CamelSagaService sagaService = endpoint.getCamelContext().hasService(CamelSagaService.class);
+ if (sagaService == null) {
+ sagaService = CamelContextHelper.findByType(endpoint.getCamelContext(), CamelSagaService.class);
+ }
+ if (sagaService == null) {
+ throw new IllegalStateException("Cannot find saga service: saga producers can only be used within a saga");
+ }
+ this.camelSagaService = sagaService;
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ String currentSaga = exchange.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION, String.class);
+ if (currentSaga == null) {
+ exchange.setException(new IllegalStateException("Current exchange is not bound to a saga context: cannot complete"));
+ callback.done(true);
+ return true;
+ }
+
+
+ camelSagaService.getSaga(currentSaga).thenApply(coordinator -> {
+ if (coordinator == null) {
+ throw new IllegalStateException("No coordinator found for saga id " + currentSaga);
+ }
+ return coordinator;
+ }).thenCompose(coordinator -> {
+ if (success) {
+ return coordinator.complete();
+ } else {
+ return coordinator.compensate();
+ }
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ exchange.setException(ex);
+ }
+ callback.done(false);
+ });
+ return false;
+ }
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/impl/saga/InMemorySagaCoordinator.java b/camel-core/src/main/java/org/apache/camel/impl/saga/InMemorySagaCoordinator.java
new file mode 100644
index 0000000..2fc5c2a
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/impl/saga/InMemorySagaCoordinator.java
@@ -0,0 +1,228 @@
+/**
+ * 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.impl.saga;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.saga.CamelSagaCoordinator;
+import org.apache.camel.saga.CamelSagaStep;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A in-memory implementation of a saga coordinator.
+ */
+public class InMemorySagaCoordinator implements CamelSagaCoordinator {
+
+ private enum Status {
+ RUNNING,
+ COMPENSATING,
+ COMPENSATED,
+ COMPLETING,
+ COMPLETED
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(InMemorySagaCoordinator.class);
+
+ private CamelContext camelContext;
+ private InMemorySagaService sagaService;
+ private String sagaId;
+ private List<CamelSagaStep> steps;
+ private Map<CamelSagaStep, Map<String, Object>> optionValues;
+ private AtomicReference<Status> currentStatus;
+
+ public InMemorySagaCoordinator(CamelContext camelContext, InMemorySagaService sagaService, String sagaId) {
+ this.camelContext = ObjectHelper.notNull(camelContext, "camelContext");
+ this.sagaService = ObjectHelper.notNull(sagaService, "sagaService");
+ this.sagaId = ObjectHelper.notNull(sagaId, "sagaId");
+ this.steps = new CopyOnWriteArrayList<>();
+ this.optionValues = new ConcurrentHashMap<>();
+ this.currentStatus = new AtomicReference<>(Status.RUNNING);
+ }
+
+ @Override
+ public String getId() {
+ return sagaId;
+ }
+
+ @Override
+ public CompletableFuture<Void> beginStep(Exchange exchange, CamelSagaStep step) {
+ this.steps.add(step);
+
+ if (!step.getOptions().isEmpty()) {
+ optionValues.putIfAbsent(step, new ConcurrentHashMap<>());
+ Map<String, Object> values = optionValues.get(step);
+ for (String option : step.getOptions().keySet()) {
+ Expression expression = step.getOptions().get(option);
+ values.put(option, expression.evaluate(exchange, Object.class));
+ }
+ }
+
+ if (step.getTimeoutInMilliseconds().isPresent()) {
+ sagaService.getExecutorService().schedule(() -> {
+ boolean doAction = currentStatus.compareAndSet(Status.RUNNING, Status.COMPENSATING);
+ if (doAction) {
+ doCompensate();
+ }
+ }, step.getTimeoutInMilliseconds().get(), TimeUnit.MILLISECONDS);
+ }
+
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> compensate() {
+ boolean doAction = currentStatus.compareAndSet(Status.RUNNING, Status.COMPENSATING);
+
+ if (doAction) {
+ doCompensate();
+ } else {
+ Status status = currentStatus.get();
+ if (status != Status.COMPENSATING && status != Status.COMPENSATED) {
+ CompletableFuture<Void> res = new CompletableFuture<>();
+ res.completeExceptionally(new IllegalStateException("Cannot compensate: status is " + status));
+ return res;
+ }
+ }
+
+ return CompletableFuture.completedFuture(null);
+ }
+
+
+ @Override
+ public CompletableFuture<Void> complete() {
+ boolean doAction = currentStatus.compareAndSet(Status.RUNNING, Status.COMPLETING);
+
+ if (doAction) {
+ doComplete();
+ } else {
+ Status status = currentStatus.get();
+ if (status != Status.COMPLETING && status != Status.COMPLETED) {
+ CompletableFuture<Void> res = new CompletableFuture<>();
+ res.completeExceptionally(new IllegalStateException("Cannot complete: status is " + status));
+ return res;
+ }
+ }
+
+ return CompletableFuture.completedFuture(null);
+ }
+
+ public CompletableFuture<Boolean> doCompensate() {
+ return doFinalize(CamelSagaStep::getCompensation, "compensation")
+ .thenApply(res -> {
+ currentStatus.set(Status.COMPENSATED);
+ return res;
+ });
+ }
+
+ public CompletableFuture<Boolean> doComplete() {
+ return doFinalize(CamelSagaStep::getCompletion, "completion")
+ .thenApply(res -> {
+ currentStatus.set(Status.COMPLETED);
+ return res;
+ });
+ }
+
+ public CompletableFuture<Boolean> doFinalize(Function<CamelSagaStep, Optional<Endpoint>> endpointExtractor, String description) {
+ CompletableFuture<Boolean> result = CompletableFuture.completedFuture(true);
+ for (CamelSagaStep step : reversed(steps)) {
+ Optional<Endpoint> endpoint = endpointExtractor.apply(step);
+ if (endpoint.isPresent()) {
+ result = result.thenCompose(prevResult ->
+ doFinalize(endpoint.get(), step, 0, description).thenApply(res -> prevResult && res));
+ }
+ }
+ return result.whenComplete((done, ex) -> {
+ if (ex != null) {
+ LOG.error("Cannot finalize " + description + " the saga", ex);
+ } else if (!done) {
+ LOG.warn("Unable to finalize " + description + " for all required steps of the saga " + sagaId);
+ }
+ });
+ }
+
+ private CompletableFuture<Boolean> doFinalize(Endpoint endpoint, CamelSagaStep step, int doneAttempts, String description) {
+ Exchange exchange = createExchange(endpoint, step);
+
+ return CompletableFuture.supplyAsync(() -> {
+ Exchange res = camelContext.createFluentProducerTemplate().to(endpoint).withExchange(exchange).send();
+ Exception ex = res.getException();
+ if (ex != null) {
+ throw new RuntimeCamelException(res.getException());
+ }
+ return true;
+ }, sagaService.getExecutorService()).exceptionally(ex -> {
+ LOG.warn("Exception thrown during " + description + " at " + endpoint.getEndpointUri()
+ + ". Attempt " + (doneAttempts + 1) + " of " + sagaService.getMaxRetryAttempts(), ex);
+ return false;
+ }).thenCompose(executed -> {
+ int currentAttempt = doneAttempts + 1;
+ if (executed) {
+ return CompletableFuture.completedFuture(true);
+ } else if (currentAttempt >= sagaService.getMaxRetryAttempts()) {
+ return CompletableFuture.completedFuture(false);
+ } else {
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ sagaService.getExecutorService().schedule(() -> {
+ doFinalize(endpoint, step, currentAttempt, description).whenComplete((res, ex) -> {
+ if (ex != null) {
+ future.completeExceptionally(ex);
+ } else {
+ future.complete(res);
+ }
+ });
+ }, sagaService.getRetryDelayInMilliseconds(), TimeUnit.MILLISECONDS);
+ return future;
+ }
+ });
+ }
+
+ private Exchange createExchange(Endpoint endpoint, CamelSagaStep step) {
+ Exchange exchange = endpoint.createExchange();
+ exchange.getIn().setHeader(Exchange.SAGA_LONG_RUNNING_ACTION, getId());
+
+ Map<String, Object> values = optionValues.get(step);
+ if (values != null) {
+ for (Map.Entry<String, Object> entry : values.entrySet()) {
+ exchange.getIn().setHeader(entry.getKey(), entry.getValue());
+ }
+ }
+ return exchange;
+ }
+
+ private <T> List<T> reversed(List<T> list) {
+ List<T> reversed = new ArrayList<>(list);
+ Collections.reverse(reversed);
+ return reversed;
+ }
+}
diff --git a/camel-core/src/main/java/org/apache/camel/impl/saga/InMemorySagaService.java b/camel-core/src/main/java/org/apache/camel/impl/saga/InMemorySagaService.java
new file mode 100644
index 0000000..6d37671
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/impl/saga/InMemorySagaService.java
@@ -0,0 +1,117 @@
+/**
+ * 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.impl.saga;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.saga.CamelSagaCoordinator;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.saga.CamelSagaStep;
+import org.apache.camel.support.ServiceSupport;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * A in-memory implementation of a saga service.
+ */
+public class InMemorySagaService extends ServiceSupport implements CamelSagaService {
+
+ public static final int DEFAULT_MAX_RETRY_ATTEMPTS = 5;
+
+ public static final long DEFAULT_RETRY_DELAY_IN_MILLISECONDS = 5000;
+
+ private CamelContext camelContext;
+
+ private Map<String, CamelSagaCoordinator> coordinators = new ConcurrentHashMap<>();
+
+ private ScheduledExecutorService executorService;
+
+ private int maxRetryAttempts = DEFAULT_MAX_RETRY_ATTEMPTS;
+
+ private long retryDelayInMilliseconds = DEFAULT_RETRY_DELAY_IN_MILLISECONDS;
+
+ @Override
+ public CompletableFuture<CamelSagaCoordinator> newSaga() {
+ ObjectHelper.notNull(camelContext, "camelContext");
+
+ String uuid = camelContext.getUuidGenerator().generateUuid();
+ CamelSagaCoordinator coordinator = new InMemorySagaCoordinator(camelContext, this, uuid);
+ coordinators.put(uuid, coordinator);
+
+ return CompletableFuture.completedFuture(coordinator);
+ }
+
+ @Override
+ public CompletableFuture<CamelSagaCoordinator> getSaga(String id) {
+ return CompletableFuture.completedFuture(coordinators.get(id));
+ }
+
+ @Override
+ public void registerStep(CamelSagaStep step) {
+ // do nothing
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ if (this.executorService == null) {
+ this.executorService = camelContext.getExecutorServiceManager()
+ .newDefaultScheduledThreadPool(this, "saga");
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (this.executorService != null) {
+ camelContext.getExecutorServiceManager().shutdownGraceful(this.executorService);
+ this.executorService = null;
+ }
+ }
+
+ public ScheduledExecutorService getExecutorService() {
+ return executorService;
+ }
+
+ @Override
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ @Override
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+
+ public int getMaxRetryAttempts() {
+ return maxRetryAttempts;
+ }
+
+ public void setMaxRetryAttempts(int maxRetryAttempts) {
+ this.maxRetryAttempts = maxRetryAttempts;
+ }
+
+ public long getRetryDelayInMilliseconds() {
+ return retryDelayInMilliseconds;
+ }
+
+ public void setRetryDelayInMilliseconds(long retryDelayInMilliseconds) {
+ this.retryDelayInMilliseconds = retryDelayInMilliseconds;
+ }
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/model/PolicyDefinition.java b/camel-core/src/main/java/org/apache/camel/model/PolicyDefinition.java
index 5471ddb..e05c537 100644
--- a/camel-core/src/main/java/org/apache/camel/model/PolicyDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/PolicyDefinition.java
@@ -94,6 +94,11 @@ public class PolicyDefinition extends OutputDefinition<PolicyDefinition> {
return false;
}
+ @Override
+ public boolean isWrappingEntireOutput() {
+ return true;
+ }
+
public String getRef() {
return ref;
}
diff --git a/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java b/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
index 594f68d..bb74e2a 100644
--- a/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
@@ -158,6 +158,18 @@ public abstract class ProcessorDefinition<Type extends ProcessorDefinition<Type>
}
/**
+ * Whether this definition is wrapping the entire output.
+ * <p/>
+ * When a definition is wrapping the entire output, the check to ensure
+ * that a route definition is empty should be done on the wrapped output.
+ *
+ * @return <tt>true</tt> when wrapping the entire output.
+ */
+ public boolean isWrappingEntireOutput() {
+ return false;
+ }
+
+ /**
* Override this in definition class and implement logic to create the processor
* based on the definition model.
*/
@@ -2696,6 +2708,17 @@ public abstract class ProcessorDefinition<Type extends ProcessorDefinition<Type>
return answer;
}
+ /**
+ * Marks this route as participating to a saga.
+ *
+ * @return the saga definition
+ */
+ public SagaDefinition saga() {
+ SagaDefinition answer = new SagaDefinition();
+ addOutput(answer);
+ return answer;
+ }
+
// Transformers
// -------------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinitionHelper.java b/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinitionHelper.java
index 451a886..15e01c1 100644
--- a/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinitionHelper.java
+++ b/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinitionHelper.java
@@ -355,7 +355,7 @@ public final class ProcessorDefinitionHelper {
return !outputs.isEmpty();
}
for (ProcessorDefinition output : outputs) {
- if (output instanceof TransactedDefinition || output instanceof PolicyDefinition) {
+ if (output.isWrappingEntireOutput()) {
// special for those as they wrap entire output, so we should just check its output
return hasOutputs(output.getOutputs(), excludeAbstract);
}
diff --git a/camel-core/src/main/java/org/apache/camel/model/RouteDefinitionHelper.java b/camel-core/src/main/java/org/apache/camel/model/RouteDefinitionHelper.java
index fccc918..81409ad 100644
--- a/camel-core/src/main/java/org/apache/camel/model/RouteDefinitionHelper.java
+++ b/camel-core/src/main/java/org/apache/camel/model/RouteDefinitionHelper.java
@@ -316,6 +316,8 @@ public final class RouteDefinitionHelper {
initInterceptors(context, route, abstracts, upper, intercepts, interceptFromDefinitions, interceptSendToEndpointDefinitions);
// then on completion
initOnCompletions(abstracts, upper, onCompletions);
+ // then sagas
+ initSagas(abstracts, lower);
// then transactions
initTransacted(abstracts, lower);
// then on exception
@@ -579,6 +581,29 @@ public final class RouteDefinitionHelper {
upper.addAll(completions);
}
+ private static void initSagas(List<ProcessorDefinition<?>> abstracts, List<ProcessorDefinition<?>> lower) {
+ SagaDefinition saga = null;
+
+ // add to correct type
+ for (ProcessorDefinition<?> type : abstracts) {
+ if (type instanceof SagaDefinition) {
+ if (saga == null) {
+ saga = (SagaDefinition) type;
+ } else {
+ throw new IllegalArgumentException("The route can only have one saga defined");
+ }
+ }
+ }
+
+ if (saga != null) {
+ // the outputs should be moved to the transacted policy
+ saga.getOutputs().addAll(lower);
+ // and add it as the single output
+ lower.clear();
+ lower.add(saga);
+ }
+ }
+
private static void initTransacted(List<ProcessorDefinition<?>> abstracts, List<ProcessorDefinition<?>> lower) {
TransactedDefinition transacted = null;
diff --git a/camel-core/src/main/java/org/apache/camel/model/SagaActionUriDefinition.java b/camel-core/src/main/java/org/apache/camel/model/SagaActionUriDefinition.java
new file mode 100644
index 0000000..39a080e
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/model/SagaActionUriDefinition.java
@@ -0,0 +1,57 @@
+/**
+ * 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.model;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlAttribute;
+
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * Allows to declare saga actions e.g. to complete or compensate a saga
+ *
+ */
+@Metadata(label = "eip,routing")
+@XmlAccessorType(XmlAccessType.FIELD)
+public class SagaActionUriDefinition {
+
+ @XmlAttribute(required = true)
+ private String uri;
+
+ public SagaActionUriDefinition() {
+ }
+
+ public SagaActionUriDefinition(String uri) {
+ this.uri = ObjectHelper.notNull(uri, "uri");
+ }
+
+ public String getUri() {
+ return uri;
+ }
+
+ public void setUri(String uri) {
+ this.uri = ObjectHelper.notNull(uri, "uri");
+ }
+
+ @Override
+ public String toString() {
+ return uri;
+ }
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/model/SagaCompletionMode.java b/camel-core/src/main/java/org/apache/camel/model/SagaCompletionMode.java
new file mode 100644
index 0000000..28f93be
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/model/SagaCompletionMode.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.model;
+
+/**
+ * Enumerates all saga completion modes.
+ */
+public enum SagaCompletionMode {
+
+ /**
+ * Complete the saga automatically as soon as the exchange is processed.
+ */
+ AUTO,
+
+ /**
+ * Complete the saga explicitly using the "saga:complete" endpoint.
+ */
+ MANUAL;
+
+ public static SagaCompletionMode defaultCompletionMode() {
+ return AUTO;
+ }
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/model/SagaDefinition.java b/camel-core/src/main/java/org/apache/camel/model/SagaDefinition.java
new file mode 100644
index 0000000..2ae0591
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/model/SagaDefinition.java
@@ -0,0 +1,347 @@
+/**
+ * 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.model;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlAttribute;
+import javax.xml.bind.annotation.XmlElement;
+import javax.xml.bind.annotation.XmlRootElement;
+import javax.xml.bind.annotation.XmlTransient;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Expression;
+import org.apache.camel.Processor;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.processor.saga.SagaProcessorBuilder;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.saga.CamelSagaStep;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.RouteContext;
+import org.apache.camel.util.CamelContextHelper;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Enables sagas on the route
+ *
+ * @version
+ */
+@Metadata(label = "eip,routing")
+@XmlRootElement(name = "saga")
+@XmlAccessorType(XmlAccessType.FIELD)
+public class SagaDefinition extends OutputDefinition<SagaDefinition> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SagaDefinition.class);
+
+ @XmlAttribute
+ @Metadata(defaultValue = "REQUIRED")
+ private SagaPropagation propagation;
+
+ @XmlAttribute
+ @Metadata(defaultValue = "AUTO")
+ private SagaCompletionMode completionMode;
+
+ @XmlAttribute
+ private Long timeoutInMilliseconds;
+
+ @XmlElement
+ private SagaActionUriDefinition compensation;
+
+ @XmlElement
+ private SagaActionUriDefinition completion;
+
+ @XmlElement(name = "option")
+ private List<SagaOptionDefinition> options;
+
+ @XmlTransient
+ private CamelSagaService sagaService; // TODO add ref for xml configuration
+
+ public SagaDefinition() {
+ }
+
+ @Override
+ public Processor createProcessor(RouteContext routeContext) throws Exception {
+ Optional<Endpoint> compensationEndpoint = Optional.ofNullable(this.compensation)
+ .map(SagaActionUriDefinition::getUri)
+ .map(routeContext::resolveEndpoint);
+
+ Optional<Endpoint> completionEndpoint = Optional.ofNullable(this.completion)
+ .map(SagaActionUriDefinition::getUri)
+ .map(routeContext::resolveEndpoint);
+
+ Map<String, Expression> optionsMap = new TreeMap<>();
+ if (this.options != null) {
+ for (SagaOptionDefinition optionDef : this.options) {
+ String optionName = optionDef.getOptionName();
+ Expression expr = optionDef.getExpression();
+ optionsMap.put(optionName, expr);
+ }
+ }
+
+ CamelSagaStep step = new CamelSagaStep(compensationEndpoint, completionEndpoint, optionsMap, Optional.ofNullable(timeoutInMilliseconds));
+
+ SagaPropagation propagation = this.propagation;
+ if (propagation == null) {
+ // default propagation mode
+ propagation = SagaPropagation.REQUIRED;
+ }
+
+ SagaCompletionMode completionMode = this.completionMode;
+ if (completionMode == null) {
+ // default completion mode
+ completionMode = SagaCompletionMode.defaultCompletionMode();
+ }
+
+ Processor childProcessor = this.createChildProcessor(routeContext, true);
+ CamelSagaService camelSagaService = findSagaService(routeContext.getCamelContext());
+
+ camelSagaService.registerStep(step);
+
+ return new SagaProcessorBuilder()
+ .camelContext(routeContext.getCamelContext())
+ .childProcessor(childProcessor)
+ .sagaService(camelSagaService)
+ .step(step)
+ .propagation(propagation)
+ .completionMode(completionMode)
+ .build();
+ }
+
+ @Override
+ public boolean isAbstract() {
+ return true;
+ }
+
+ @Override
+ public boolean isTopLevelOnly() {
+ return true;
+ }
+
+ @Override
+ public boolean isWrappingEntireOutput() {
+ return true;
+ }
+
+ @Override
+ public String getLabel() {
+ String desc = description();
+ if (ObjectHelper.isEmpty(desc)) {
+ return "saga";
+ } else {
+ return "saga[" + desc + "]";
+ }
+ }
+
+ @Override
+ public String toString() {
+ String desc = description();
+ if (ObjectHelper.isEmpty(desc)) {
+ return "Saga -> [" + outputs + "]";
+ } else {
+ return "Saga[" + desc + "] -> [" + outputs + "]";
+ }
+ }
+
+ // Properties
+
+
+ public SagaActionUriDefinition getCompensation() {
+ return compensation;
+ }
+
+ /**
+ * The compensation endpoint URI that must be called to compensate all changes done in the route.
+ * The route corresponding to the compensation URI must perform compensation and complete without error.
+ *
+ * If errors occur during compensation, the saga service may call again the compensation URI to retry.
+ */
+ public void setCompensation(SagaActionUriDefinition compensation) {
+ this.compensation = compensation;
+ }
+
+ public SagaActionUriDefinition getCompletion() {
+ return completion;
+ }
+
+ /**
+ * The completion endpoint URI that will be called when the Saga is completed successfully.
+ * The route corresponding to the completion URI must perform completion tasks and terminate without error.
+ *
+ * If errors occur during completion, the saga service may call again the completion URI to retry.
+ */
+ public void setCompletion(SagaActionUriDefinition completion) {
+ this.completion = completion;
+ }
+
+ public SagaPropagation getPropagation() {
+ return propagation;
+ }
+
+ /**
+ * Set the Saga propagation mode (REQUIRED, REQUIRES_NEW, MANDATORY, SUPPORTS, NOT_SUPPORTED, NEVER).
+ */
+ public void setPropagation(SagaPropagation propagation) {
+ this.propagation = propagation;
+ }
+
+ public SagaCompletionMode getCompletionMode() {
+ return completionMode;
+ }
+
+ /**
+ * Determine how the saga should be considered complete. When set to AUTO, the saga is completed when the exchange that
+ * initiates the saga is processed successfully, or compensated when it completes exceptionally.
+ *
+ * When set to MANUAL, the user must complete or compensate the saga using the "saga:complete" or "saga:compensate" endpoints.
+ */
+ public void setCompletionMode(SagaCompletionMode completionMode) {
+ this.completionMode = completionMode;
+ }
+
+ public CamelSagaService getSagaService() {
+ return sagaService;
+ }
+
+ public void setSagaService(CamelSagaService sagaService) {
+ this.sagaService = sagaService;
+ }
+
+ public List<SagaOptionDefinition> getOptions() {
+ return options;
+ }
+
+ /**
+ * Allows to save properties of the current exchange in order to re-use them in a compensation/completion callback route.
+ * Options are usually helpful e.g. to store and retrieve identifiers of objects that should be deleted in compensating actions.
+ *
+ * Option values will be transformed into input headers of the compensation/completion exchange.
+ */
+ public void setOptions(List<SagaOptionDefinition> options) {
+ this.options = options;
+ }
+
+ public Long getTimeoutInMilliseconds() {
+ return timeoutInMilliseconds;
+ }
+
+ /**
+ * Set the maximum amount of time for the Saga. After the timeout is expired, the saga will be compensated
+ * automatically (unless a different decision has been taken in the meantime).
+ */
+ public void setTimeoutInMilliseconds(Long timeoutInMilliseconds) {
+ this.timeoutInMilliseconds = timeoutInMilliseconds;
+ }
+
+ private void addOption(String option, Expression expression) {
+ if (this.options == null) {
+ this.options = new ArrayList<>();
+ }
+ this.options.add(new SagaOptionDefinition(option, expression));
+ }
+
+ // Builders
+
+ public SagaDefinition compensation(String compensation) {
+ if (this.compensation != null) {
+ throw new IllegalStateException("Compensation has already been set");
+ }
+ this.compensation = new SagaActionUriDefinition(compensation);
+ return this;
+ }
+
+ public SagaDefinition completion(String completion) {
+ if (this.completion != null) {
+ throw new IllegalStateException("Completion has already been set");
+ }
+ this.completion = new SagaActionUriDefinition(completion);
+ return this;
+ }
+
+ public SagaDefinition propagation(SagaPropagation propagation) {
+ setPropagation(propagation);
+ return this;
+ }
+
+ public SagaDefinition sagaService(CamelSagaService sagaService) {
+ setSagaService(sagaService);
+ return this;
+ }
+
+ public SagaDefinition completionMode(SagaCompletionMode completionMode) {
+ setCompletionMode(completionMode);
+ return this;
+ }
+
+ public SagaDefinition option(String option, Expression expression) {
+ addOption(option, expression);
+ return this;
+ }
+
+ public SagaDefinition timeout(long timeout, TimeUnit unit) {
+ setTimeoutInMilliseconds(unit.toMillis(timeout));
+ return this;
+ }
+
+ // Utils
+
+ protected CamelSagaService findSagaService(CamelContext context) {
+ CamelSagaService sagaService = getSagaService();
+ if (sagaService != null) {
+ return sagaService;
+ }
+
+ sagaService = context.hasService(CamelSagaService.class);
+ if (sagaService != null) {
+ return sagaService;
+ }
+
+ sagaService = CamelContextHelper.findByType(context, CamelSagaService.class);
+ if (sagaService != null) {
+ return sagaService;
+ }
+
+ throw new RuntimeCamelException("Cannot find a CamelSagaService");
+ }
+
+ protected String description() {
+ StringBuilder desc = new StringBuilder();
+ addField(desc, "compensation", compensation);
+ addField(desc, "completion", completion);
+ addField(desc, "propagation", propagation);
+ return desc.toString();
+ }
+
+ private void addField(StringBuilder builder, String key, Object value) {
+ if (value == null) {
+ return;
+ }
+ if (builder.length() > 0) {
+ builder.append(',');
+ }
+ builder.append(key).append(':').append(value);
+ }
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/model/SagaOptionDefinition.java b/camel-core/src/main/java/org/apache/camel/model/SagaOptionDefinition.java
new file mode 100644
index 0000000..666e17b
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/model/SagaOptionDefinition.java
@@ -0,0 +1,79 @@
+/**
+ * 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.model;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlAttribute;
+import javax.xml.bind.annotation.XmlElementRef;
+import javax.xml.bind.annotation.XmlRootElement;
+
+import org.apache.camel.Expression;
+import org.apache.camel.model.language.ExpressionDefinition;
+import org.apache.camel.spi.Metadata;
+
+/**
+ * Allows to declare options on sagas
+ *
+ */
+@Metadata(label = "eip,routing")
+@XmlAccessorType(XmlAccessType.FIELD)
+public class SagaOptionDefinition {
+
+ @XmlAttribute(required = true)
+ private String optionName;
+
+ @XmlElementRef
+ private ExpressionDefinition expression;
+
+
+ public SagaOptionDefinition() {
+ }
+
+ public SagaOptionDefinition(String optionName, Expression expression) {
+ setOptionName(optionName);
+ setExpression(ExpressionNodeHelper.toExpressionDefinition(expression));
+ }
+
+ @Override
+ public String toString() {
+ return "option:" + getOptionName() + "=" + getExpression();
+ }
+
+ /**
+ * Name of the option. It identifies the name of the header where the value of the expression will be stored when the
+ * compensation or completion routes will be called.
+ */
+ public void setOptionName(String optionName) {
+ this.optionName = optionName;
+ }
+
+ public String getOptionName() {
+ return optionName;
+ }
+
+ public ExpressionDefinition getExpression() {
+ return expression;
+ }
+
+ /**
+ * The expression to be used to determine the value of the option.
+ */
+ public void setExpression(ExpressionDefinition expression) {
+ this.expression = expression;
+ }
+}
diff --git a/camel-core/src/main/java/org/apache/camel/model/SagaPropagation.java b/camel-core/src/main/java/org/apache/camel/model/SagaPropagation.java
new file mode 100644
index 0000000..8a16d95
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/model/SagaPropagation.java
@@ -0,0 +1,54 @@
+/**
+ * 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.model;
+
+/**
+ * Enumerates all saga propagation modes.
+ */
+public enum SagaPropagation {
+
+ /**
+ * Join the existing saga or create a new one if it does not exist.
+ */
+ REQUIRED,
+
+ /**
+ * Always create a new saga. Suspend the old saga and resume it when the new one terminates.
+ */
+ REQUIRES_NEW,
+
+ /**
+ * A saga must be already present. The existing saga is joined.
+ */
+ MANDATORY,
+
+ /**
+ * If a saga already exists, then join it.
+ */
+ SUPPORTS,
+
+ /**
+ * If a saga already exists, it is suspended and resumed when the current block completes.
+ */
+ NOT_SUPPORTED,
+
+ /**
+ * The current block must never be invoked within a saga.
+ */
+ NEVER
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/model/TransactedDefinition.java b/camel-core/src/main/java/org/apache/camel/model/TransactedDefinition.java
index 29e5565..eb80758 100644
--- a/camel-core/src/main/java/org/apache/camel/model/TransactedDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/TransactedDefinition.java
@@ -112,6 +112,11 @@ public class TransactedDefinition extends OutputDefinition<TransactedDefinition>
return true;
}
+ @Override
+ public boolean isWrappingEntireOutput() {
+ return true;
+ }
+
public String getRef() {
return ref;
}
diff --git a/camel-core/src/main/java/org/apache/camel/processor/saga/MandatorySagaProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/saga/MandatorySagaProcessor.java
new file mode 100644
index 0000000..53c6577
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/processor/saga/MandatorySagaProcessor.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.saga;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.model.SagaCompletionMode;
+import org.apache.camel.saga.CamelSagaCoordinator;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.saga.CamelSagaStep;
+
+/**
+ * Saga processor implementing the MANDATORY propagation mode.
+ */
+public class MandatorySagaProcessor extends SagaProcessor {
+
+ public MandatorySagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService, SagaCompletionMode completionMode, CamelSagaStep step) {
+ super(camelContext, childProcessor, sagaService, completionMode, step);
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ getCurrentSagaCoordinator(exchange).whenComplete((coordinator, ex) -> ifNotException(ex, exchange, callback, () -> {
+ if (coordinator == null) {
+ exchange.setException(new CamelExchangeException("Exchange is not part of a saga", exchange));
+ callback.done(false);
+ } else {
+ coordinator.beginStep(exchange, step).whenComplete((done, ex2) -> ifNotException(ex2, exchange, callback, () -> {
+ super.process(exchange, doneSync -> callback.done(false));
+ }));
+ }
+ }));
+ return false;
+ }
+}
diff --git a/camel-core/src/main/java/org/apache/camel/processor/saga/NeverSagaProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/saga/NeverSagaProcessor.java
new file mode 100644
index 0000000..7458425
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/processor/saga/NeverSagaProcessor.java
@@ -0,0 +1,56 @@
+/**
+ * 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.saga;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.model.SagaCompletionMode;
+import org.apache.camel.saga.CamelSagaCoordinator;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.saga.CamelSagaStep;
+
+/**
+ * Saga processor implementing the NEVER propagation mode.
+ */
+public class NeverSagaProcessor extends SagaProcessor {
+
+ public NeverSagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService, SagaCompletionMode completionMode, CamelSagaStep step) {
+ super(camelContext, childProcessor, sagaService, completionMode, step);
+ if (!step.isEmpty()) {
+ throw new IllegalArgumentException("Saga configuration is not allowed when propagation is set to NEVER");
+ }
+ if (completionMode != null && completionMode != SagaCompletionMode.defaultCompletionMode()) {
+ throw new IllegalArgumentException("CompletionMode cannot be specified when propagation is NEVER");
+ }
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ getCurrentSagaCoordinator(exchange).whenComplete((coordinator, ex) -> ifNotException(ex, exchange, callback, () -> {
+ if (coordinator != null) {
+ exchange.setException(new CamelExchangeException("Route cannot handle exchanges that are joining a saga", exchange));
+ callback.done(false);
+ } else {
+ super.process(exchange, doneSync -> callback.done(false));
+ }
+ }));
+ return false;
+ }
+}
diff --git a/camel-core/src/main/java/org/apache/camel/processor/saga/NotSupportedSagaProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/saga/NotSupportedSagaProcessor.java
new file mode 100644
index 0000000..a9d5050
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/processor/saga/NotSupportedSagaProcessor.java
@@ -0,0 +1,56 @@
+/**
+ * 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.saga;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.model.SagaCompletionMode;
+import org.apache.camel.saga.CamelSagaCoordinator;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.saga.CamelSagaStep;
+
+/**
+ * Saga processor implementing the NOT_SUPPORTED propagation mode.
+ */
+public class NotSupportedSagaProcessor extends SagaProcessor {
+
+ public NotSupportedSagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService, SagaCompletionMode completionMode, CamelSagaStep step) {
+ super(camelContext, childProcessor, sagaService, completionMode, step);
+ if (!step.isEmpty()) {
+ throw new IllegalArgumentException("Saga configuration is not allowed when propagation is set to NOT_SUPPORTED");
+ }
+ if (completionMode != null && completionMode != SagaCompletionMode.defaultCompletionMode()) {
+ throw new IllegalArgumentException("CompletionMode cannot be specified when propagation is NOT_SUPPORTED");
+ }
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ getCurrentSagaCoordinator(exchange).whenComplete((coordinator, ex) -> ifNotException(ex, exchange, callback, () -> {
+ setCurrentSagaCoordinator(exchange, null);
+
+ super.process(exchange, doneSync -> {
+ // Restore existing coordinator
+ setCurrentSagaCoordinator(exchange, coordinator);
+ callback.done(false);
+ });
+ }));
+ return false;
+ }
+}
diff --git a/camel-core/src/main/java/org/apache/camel/processor/saga/RequiredSagaProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/saga/RequiredSagaProcessor.java
new file mode 100644
index 0000000..7b77642
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/processor/saga/RequiredSagaProcessor.java
@@ -0,0 +1,69 @@
+/**
+ * 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.saga;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.model.SagaCompletionMode;
+import org.apache.camel.saga.CamelSagaCoordinator;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.saga.CamelSagaStep;
+
+/**
+ * Saga processor implementing the REQUIRED propagation mode.
+ */
+public class RequiredSagaProcessor extends SagaProcessor {
+
+ public RequiredSagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService, SagaCompletionMode completionMode, CamelSagaStep step) {
+ super(camelContext, childProcessor, sagaService, completionMode, step);
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ getCurrentSagaCoordinator(exchange).whenComplete((existingCoordinator, ex) -> ifNotException(ex, exchange, callback, () -> {
+ CompletableFuture<CamelSagaCoordinator> coordinatorFuture;
+ final boolean inheritedCoordinator;
+ if (existingCoordinator != null) {
+ coordinatorFuture = CompletableFuture.completedFuture(existingCoordinator);
+ inheritedCoordinator = true;
+ } else {
+ coordinatorFuture = sagaService.newSaga();
+ inheritedCoordinator = false;
+ }
+
+ coordinatorFuture.whenComplete((coordinator, ex2) -> ifNotException(ex2, exchange, callback, () -> {
+ setCurrentSagaCoordinator(exchange, coordinator);
+ coordinator.beginStep(exchange, step).whenComplete((done, ex3) -> ifNotException(ex3, exchange, callback, () -> {
+ super.process(exchange, doneSync -> {
+ if (!inheritedCoordinator) {
+ // Saga starts and ends here
+ handleSagaCompletion(exchange, coordinator, null, callback);
+ } else {
+ callback.done(false);
+ }
+ });
+ }));
+ }));
+ }));
+
+ return false;
+ }
+}
diff --git a/camel-core/src/main/java/org/apache/camel/processor/saga/RequiresNewSagaProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/saga/RequiresNewSagaProcessor.java
new file mode 100644
index 0000000..a87fb55
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/processor/saga/RequiresNewSagaProcessor.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.saga;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.model.SagaCompletionMode;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.saga.CamelSagaStep;
+
+/**
+ * Saga processor implementing the REQUIRES_NEW propagation mode.
+ */
+public class RequiresNewSagaProcessor extends SagaProcessor {
+
+ public RequiresNewSagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService, SagaCompletionMode completionMode, CamelSagaStep step) {
+ super(camelContext, childProcessor, sagaService, completionMode, step);
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ getCurrentSagaCoordinator(exchange).whenComplete((existingCoordinator, ex) -> ifNotException(ex, exchange, callback, () ->
+ sagaService.newSaga().whenComplete((newCoordinator, ex2) -> ifNotException(ex2, exchange, callback, () -> {
+ setCurrentSagaCoordinator(exchange, newCoordinator);
+
+ newCoordinator.beginStep(exchange, step).whenComplete((done, ex3) -> ifNotException(ex3, exchange, callback, () -> {
+ // Always finalizes the saga
+ super.process(exchange, doneSync -> handleSagaCompletion(exchange, newCoordinator, existingCoordinator, callback));
+ }));
+
+ }))));
+
+ return false;
+ }
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java
new file mode 100644
index 0000000..ccd239c
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/processor/saga/SagaProcessor.java
@@ -0,0 +1,109 @@
+/**
+ * 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.saga;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.model.SagaCompletionMode;
+import org.apache.camel.processor.DelegateAsyncProcessor;
+import org.apache.camel.saga.CamelSagaCoordinator;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.saga.CamelSagaStep;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * Processor for handling sagas.
+ */
+public abstract class SagaProcessor extends DelegateAsyncProcessor {
+
+ protected CamelContext camelContext;
+
+ protected CamelSagaService sagaService;
+
+ protected CamelSagaStep step;
+
+ protected SagaCompletionMode completionMode;
+
+ public SagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService, SagaCompletionMode completionMode, CamelSagaStep step) {
+ super(ObjectHelper.notNull(childProcessor, "childProcessor"));
+ this.camelContext = ObjectHelper.notNull(camelContext, "camelContext");
+ this.sagaService = ObjectHelper.notNull(sagaService, "sagaService");
+ this.completionMode = ObjectHelper.notNull(completionMode, "completionMode");
+ this.step = ObjectHelper.notNull(step, "step");
+ }
+
+ protected CompletableFuture<CamelSagaCoordinator> getCurrentSagaCoordinator(Exchange exchange) {
+ String currentSaga = exchange.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION, String.class);
+ if (currentSaga != null) {
+ return sagaService.getSaga(currentSaga);
+ }
+
+ return CompletableFuture.completedFuture(null);
+ }
+
+ protected void setCurrentSagaCoordinator(Exchange exchange, CamelSagaCoordinator coordinator) {
+ if (coordinator != null) {
+ exchange.getIn().setHeader(Exchange.SAGA_LONG_RUNNING_ACTION, coordinator.getId());
+ } else {
+ exchange.getIn().removeHeader(Exchange.SAGA_LONG_RUNNING_ACTION);
+ }
+ }
+
+ protected void handleSagaCompletion(Exchange exchange, CamelSagaCoordinator coordinator, CamelSagaCoordinator previousCoordinator, AsyncCallback callback) {
+ if (this.completionMode == SagaCompletionMode.AUTO) {
+ if (exchange.getException() != null) {
+ coordinator.compensate().whenComplete((done, ex) -> ifNotException(ex, exchange, callback, () -> {
+ setCurrentSagaCoordinator(exchange, previousCoordinator);
+ callback.done(false);
+ }));
+ } else {
+ coordinator.complete().whenComplete((done, ex) -> ifNotException(ex, exchange, callback, () -> {
+ setCurrentSagaCoordinator(exchange, previousCoordinator);
+ callback.done(false);
+ }));
+ }
+ } else if (this.completionMode == SagaCompletionMode.MANUAL) {
+ // Completion will be handled manually by the user
+ callback.done(false);
+ } else {
+ throw new IllegalStateException("Unsupported completion mode: " + this.completionMode);
+ }
+ }
+
+ public CamelSagaService getSagaService() {
+ return sagaService;
+ }
+
+ @Override
+ public String toString() {
+ return "saga";
+ }
+
+ protected void ifNotException(Throwable ex, Exchange exchange, AsyncCallback callback, Runnable code) {
+ if (ex != null) {
+ exchange.setException(ex);
+ callback.done(false);
+ } else {
+ code.run();
+ }
+ }
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/processor/saga/SagaProcessorBuilder.java b/camel-core/src/main/java/org/apache/camel/processor/saga/SagaProcessorBuilder.java
new file mode 100644
index 0000000..d975c7c
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/processor/saga/SagaProcessorBuilder.java
@@ -0,0 +1,99 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor.saga;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Processor;
+import org.apache.camel.model.SagaCompletionMode;
+import org.apache.camel.model.SagaPropagation;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.saga.CamelSagaStep;
+
+/**
+ * Builder of Saga processors.
+ */
+public class SagaProcessorBuilder {
+
+ private CamelContext camelContext;
+
+ private Processor childProcessor;
+
+ private CamelSagaService sagaService;
+
+ private CamelSagaStep step;
+
+ private SagaPropagation propagation;
+
+ private SagaCompletionMode completionMode;
+
+ public SagaProcessorBuilder() {
+ }
+
+ public SagaProcessorBuilder camelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ return this;
+ }
+
+ public SagaProcessorBuilder childProcessor(Processor childProcessor) {
+ this.childProcessor = childProcessor;
+ return this;
+ }
+
+ public SagaProcessorBuilder sagaService(CamelSagaService sagaService) {
+ this.sagaService = sagaService;
+ return this;
+ }
+
+ public SagaProcessorBuilder step(CamelSagaStep step) {
+ this.step = step;
+ return this;
+ }
+
+ public SagaProcessorBuilder propagation(SagaPropagation propagation) {
+ this.propagation = propagation;
+ return this;
+ }
+
+ public SagaProcessorBuilder completionMode(SagaCompletionMode completionMode) {
+ this.completionMode = completionMode;
+ return this;
+ }
+
+ public SagaProcessor build() {
+ if (propagation == null) {
+ throw new IllegalStateException("A propagation mode has not been set");
+ }
+
+ switch (propagation) {
+ case REQUIRED:
+ return new RequiredSagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+ case REQUIRES_NEW:
+ return new RequiresNewSagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+ case SUPPORTS:
+ return new SupportsSagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+ case NOT_SUPPORTED:
+ return new NotSupportedSagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+ case NEVER:
+ return new NeverSagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+ case MANDATORY:
+ return new MandatorySagaProcessor(camelContext, childProcessor, sagaService, completionMode, step);
+ default:
+ throw new IllegalStateException("Unsupported propagation mode: " + propagation);
+ }
+ }
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/processor/saga/SupportsSagaProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/saga/SupportsSagaProcessor.java
new file mode 100644
index 0000000..94773a1
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/processor/saga/SupportsSagaProcessor.java
@@ -0,0 +1,54 @@
+/**
+ * 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.saga;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.model.SagaCompletionMode;
+import org.apache.camel.saga.CamelSagaCoordinator;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.saga.CamelSagaStep;
+
+/**
+ * Saga processor implementing the SUPPORTS propagation mode.
+ */
+public class SupportsSagaProcessor extends SagaProcessor {
+
+ public SupportsSagaProcessor(CamelContext camelContext, Processor childProcessor, CamelSagaService sagaService, SagaCompletionMode completionMode, CamelSagaStep step) {
+ super(camelContext, childProcessor, sagaService, completionMode, step);
+ if (completionMode != null && completionMode != SagaCompletionMode.defaultCompletionMode()) {
+ throw new IllegalArgumentException("CompletionMode cannot be specified when propagation is SUPPORTS");
+ }
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ getCurrentSagaCoordinator(exchange).whenComplete((coordinator, ex) -> ifNotException(ex, exchange, callback, () -> {
+ if (coordinator != null) {
+ coordinator.beginStep(exchange, step).whenComplete((done, ex2) -> ifNotException(ex2, exchange, callback, () -> {
+ // Never completes the saga
+ super.process(exchange, doneSync -> callback.done(false));
+ }));
+ } else {
+ super.process(exchange, doneSync -> callback.done(false));
+ }
+ }));
+ return false;
+ }
+}
diff --git a/camel-core/src/main/java/org/apache/camel/saga/CamelSagaCoordinator.java b/camel-core/src/main/java/org/apache/camel/saga/CamelSagaCoordinator.java
new file mode 100644
index 0000000..c71a1a8
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/saga/CamelSagaCoordinator.java
@@ -0,0 +1,36 @@
+/**
+ * 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.saga;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.HasId;
+
+
+/**
+ * A saga coordinator can be used to register compensators and take the final decision on the saga: compensate or complete (successfully).
+ */
+public interface CamelSagaCoordinator extends HasId {
+
+ CompletableFuture<Void> beginStep(Exchange exchange, CamelSagaStep step);
+
+ CompletableFuture<Void> compensate();
+
+ CompletableFuture<Void> complete();
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/saga/CamelSagaService.java b/camel-core/src/main/java/org/apache/camel/saga/CamelSagaService.java
new file mode 100644
index 0000000..9547129
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/saga/CamelSagaService.java
@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.saga;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.Service;
+
+/**
+ * A Camel saga service is a factory of saga coordinators.
+ */
+public interface CamelSagaService extends Service, CamelContextAware {
+
+ CompletableFuture<CamelSagaCoordinator> newSaga();
+
+ CompletableFuture<CamelSagaCoordinator> getSaga(String id);
+
+ void registerStep(CamelSagaStep step);
+
+}
diff --git a/camel-core/src/main/java/org/apache/camel/saga/CamelSagaStep.java b/camel-core/src/main/java/org/apache/camel/saga/CamelSagaStep.java
new file mode 100644
index 0000000..715d3a8
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/saga/CamelSagaStep.java
@@ -0,0 +1,65 @@
+/**
+ * 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.saga;
+
+import java.util.Map;
+import java.util.Optional;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Expression;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * Defines the configuration of a saga step.
+ */
+public class CamelSagaStep {
+
+ private Optional<Endpoint> compensation;
+
+ private Optional<Endpoint> completion;
+
+ private Map<String, Expression> options;
+
+ private Optional<Long> timeoutInMilliseconds;
+
+ public CamelSagaStep(Optional<Endpoint> compensation, Optional<Endpoint> completion, Map<String, Expression> options, Optional<Long> timeoutInMilliseconds) {
+ this.compensation = ObjectHelper.notNull(compensation, "compensation");
+ this.completion = ObjectHelper.notNull(completion, "completionCallbacks");
+ this.options = ObjectHelper.notNull(options, "options");
+ this.timeoutInMilliseconds = ObjectHelper.notNull(timeoutInMilliseconds, "timeoutInMilliseconds");
+ }
+
+ public Optional<Endpoint> getCompensation() {
+ return compensation;
+ }
+
+ public Optional<Endpoint> getCompletion() {
+ return completion;
+ }
+
+ public Map<String, Expression> getOptions() {
+ return options;
+ }
+
+ public Optional<Long> getTimeoutInMilliseconds() {
+ return timeoutInMilliseconds;
+ }
+
+ public boolean isEmpty() {
+ return !compensation.isPresent() && !completion.isPresent() && options.isEmpty() && !timeoutInMilliseconds.isPresent();
+ }
+}
diff --git a/camel-core/src/main/resources/META-INF/services/org/apache/camel/component/saga b/camel-core/src/main/resources/META-INF/services/org/apache/camel/component/saga
new file mode 100644
index 0000000..0a4dd38
--- /dev/null
+++ b/camel-core/src/main/resources/META-INF/services/org/apache/camel/component/saga
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+class=org.apache.camel.component.saga.SagaComponent
diff --git a/camel-core/src/main/resources/org/apache/camel/model/jaxb.index b/camel-core/src/main/resources/org/apache/camel/model/jaxb.index
index b1b877d..793f23b 100644
--- a/camel-core/src/main/resources/org/apache/camel/model/jaxb.index
+++ b/camel-core/src/main/resources/org/apache/camel/model/jaxb.index
@@ -76,6 +76,7 @@ RouteDefinition
RouteContextRefDefinition
RoutesDefinition
RoutingSlipDefinition
+SagaDefinition
SamplingDefinition
ScriptDefinition
SetBodyDefinition
diff --git a/camel-core/src/test/java/org/apache/camel/component/saga/SagaComponentTest.java b/camel-core/src/test/java/org/apache/camel/component/saga/SagaComponentTest.java
new file mode 100644
index 0000000..099e49a
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/component/saga/SagaComponentTest.java
@@ -0,0 +1,132 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.saga;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.saga.InMemorySagaService;
+import org.apache.camel.model.SagaCompletionMode;
+import org.junit.Assert;
+
+/**
+ *
+ */
+public class SagaComponentTest extends ContextTestSupport {
+
+ public void testManualCompletion() throws InterruptedException {
+ MockEndpoint completed = getMockEndpoint("mock:completed");
+ completed.expectedMessageCount(1);
+
+ template.sendBody("direct:manual-workflow", "manual-complete");
+
+ completed.assertIsSatisfied();
+ }
+
+ public void testManualCompletionIsNotTriggeredAutomatically() throws InterruptedException {
+ MockEndpoint completed = getMockEndpoint("mock:completed");
+ completed.expectedMessageCount(1);
+ completed.setResultWaitTime(1000);
+
+ template.sendBody("direct:manual-workflow", "do-not-complete");
+
+ completed.assertIsNotSatisfied();
+ }
+
+ public void testManualCompensationIsTriggeredOnly() throws InterruptedException {
+ MockEndpoint completed = getMockEndpoint("mock:completed");
+ completed.expectedMessageCount(1);
+ completed.setResultWaitTime(1000);
+
+ MockEndpoint compensated = getMockEndpoint("mock:compensated");
+ compensated.expectedMessageCount(1);
+
+ template.sendBody("direct:manual-workflow", "manual-compensate");
+
+ completed.assertIsNotSatisfied();
+ compensated.assertIsSatisfied();
+ }
+
+ public void testAutoCompletion() throws InterruptedException {
+ MockEndpoint completed = getMockEndpoint("mock:completed");
+ completed.expectedMessageCount(1);
+
+ template.sendBody("direct:auto-workflow", "auto-complete");
+
+ completed.assertIsSatisfied();
+ }
+
+ public void testAutoCompensationIsTriggeredOnly() throws InterruptedException {
+ MockEndpoint completed = getMockEndpoint("mock:completed");
+ completed.expectedMessageCount(1);
+ completed.setResultWaitTime(1000);
+
+ MockEndpoint compensated = getMockEndpoint("mock:compensated");
+ compensated.expectedMessageCount(1);
+
+ try {
+ template.sendBody("direct:auto-workflow", "auto-compensate");
+ Assert.fail("Should throw an exception");
+ } catch (Exception ex) {
+ // OK
+ }
+
+ completed.assertIsNotSatisfied();
+ compensated.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ context.addService(new InMemorySagaService());
+
+ // Manual complete/compensate
+ from("direct:manual-workflow")
+ .saga()
+ .compensation("mock:compensated")
+ .completion("mock:completed")
+ .completionMode(SagaCompletionMode.MANUAL)
+ .to("seda:async");
+
+ from("seda:async")
+ .choice()
+ .when(body().isEqualTo(constant("manual-complete")))
+ .to("saga:complete")
+ .when(body().isEqualTo(constant("manual-compensate")))
+ .to("saga:compensate")
+ .end();
+
+
+ // Auto complete/compensate
+ from("direct:auto-workflow")
+ .saga()
+ .completion("mock:completed")
+ .compensation("mock:compensated")
+ .choice()
+ .when(body().isEqualTo(constant("auto-compensate")))
+ .process(x -> {
+ throw new RuntimeException("mock exception");
+ })
+ .end()
+ .to("seda:async");
+ }
+ };
+ }
+}
diff --git a/camel-core/src/test/java/org/apache/camel/model/XmlParseTest.java b/camel-core/src/test/java/org/apache/camel/model/XmlParseTest.java
index 1426b72..ff90457 100644
--- a/camel-core/src/test/java/org/apache/camel/model/XmlParseTest.java
+++ b/camel-core/src/test/java/org/apache/camel/model/XmlParseTest.java
@@ -94,6 +94,14 @@ public class XmlParseTest extends XmlTestSupport {
assertChildTo(route, "mock:end", 1);
}
+ public void testParseSagaXml() throws Exception {
+ RouteDefinition route = assertOneRoute("saga.xml");
+ assertFrom(route, "direct:start");
+ SagaDefinition node = assertNthProcessorInstanceOf(SagaDefinition.class, route, 0);
+ assertNotNull(node.getCompensation());
+ assertChildTo(route, "mock:end", 2);
+ }
+
public void testParseScriptXml() throws Exception {
RouteDefinition route = assertOneRoute("script.xml");
assertFrom(route, "direct:start");
diff --git a/camel-core/src/test/java/org/apache/camel/processor/SagaFailuresTest.java b/camel-core/src/test/java/org/apache/camel/processor/SagaFailuresTest.java
new file mode 100644
index 0000000..03c9104
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/SagaFailuresTest.java
@@ -0,0 +1,131 @@
+/**
+ * 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 java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.saga.InMemorySagaService;
+
+public class SagaFailuresTest extends ContextTestSupport {
+
+ private AtomicInteger maxFailures;
+
+ public void testCompensationAfterFailures() throws Exception {
+ maxFailures = new AtomicInteger(2);
+
+ MockEndpoint compensate = getMockEndpoint("mock:compensate");
+ compensate.expectedMessageCount(1);
+
+ sendBody("direct:saga-compensate", "hello");
+
+ compensate.assertIsSatisfied();
+ }
+
+ public void testNoCompensationAfterMaxFailures() throws Exception {
+ maxFailures = new AtomicInteger(3);
+
+ MockEndpoint compensate = getMockEndpoint("mock:compensate");
+ compensate.expectedMessageCount(1);
+ compensate.setResultWaitTime(200);
+
+ sendBody("direct:saga-compensate", "hello");
+
+ compensate.assertIsNotSatisfied();
+ }
+
+ public void testCompletionAfterFailures() throws Exception {
+ maxFailures = new AtomicInteger(2);
+
+ MockEndpoint complete = getMockEndpoint("mock:complete");
+ complete.expectedMessageCount(1);
+
+ MockEndpoint end = getMockEndpoint("mock:end");
+ end.expectedBodiesReceived("hello");
+
+ sendBody("direct:saga-complete", "hello");
+
+ complete.assertIsSatisfied();
+ end.assertIsSatisfied();
+ }
+
+ public void testNoCompletionAfterMaxFailures() throws Exception {
+ maxFailures = new AtomicInteger(3);
+
+ MockEndpoint complete = getMockEndpoint("mock:complete");
+ complete.expectedMessageCount(1);
+ complete.setResultWaitTime(200);
+
+ MockEndpoint end = getMockEndpoint("mock:end");
+ end.expectedBodiesReceived("hello");
+
+ sendBody("direct:saga-complete", "hello");
+
+ complete.assertIsNotSatisfied();
+ end.assertIsSatisfied();
+ }
+
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ InMemorySagaService sagaService = new InMemorySagaService();
+ sagaService.setMaxRetryAttempts(3);
+ sagaService.setRetryDelayInMilliseconds(20);
+ context.addService(sagaService);
+
+ from("direct:saga-compensate")
+ .saga()
+ .compensation("direct:compensate")
+ .process(x -> {
+ throw new RuntimeException("fail");
+ });
+
+ from("direct:saga-complete")
+ .saga()
+ .completion("direct:complete")
+ .to("mock:end");
+
+ from("direct:compensate")
+ .process(x -> {
+ int current = maxFailures.decrementAndGet();
+ if (current >= 0) {
+ throw new RuntimeException("compensation failure");
+ }
+ })
+ .to("mock:compensate");
+
+ from("direct:complete")
+ .process(x -> {
+ int current = maxFailures.decrementAndGet();
+ if (current >= 0) {
+ throw new RuntimeException("completion failure");
+ }
+ })
+ .to("mock:complete");
+
+ }
+ };
+ }
+
+}
diff --git a/camel-core/src/test/java/org/apache/camel/processor/SagaOptionsTest.java b/camel-core/src/test/java/org/apache/camel/processor/SagaOptionsTest.java
new file mode 100644
index 0000000..0551904
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/SagaOptionsTest.java
@@ -0,0 +1,89 @@
+/**
+ * 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.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.saga.InMemorySagaService;
+import org.junit.Assert;
+
+public class SagaOptionsTest extends ContextTestSupport {
+
+
+ public void testHeaderForwardedToComplete() throws Exception {
+
+ MockEndpoint complete = getMockEndpoint("mock:complete");
+ complete.expectedMessageCount(1);
+ complete.expectedHeaderReceived("id", "myheader");
+ complete.expectedHeaderReceived("name", "Nicola");
+ complete.expectedMessagesMatches(ex -> ex.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION) != null);
+
+ template.sendBodyAndHeader("direct:workflow", "Hello", "myname", "Nicola");
+
+ complete.assertIsSatisfied();
+ }
+
+ public void testHeaderForwardedToCompensate() throws Exception {
+
+ MockEndpoint compensate = getMockEndpoint("mock:compensate");
+ compensate.expectedMessageCount(1);
+ compensate.expectedHeaderReceived("id", "myheader");
+ compensate.expectedHeaderReceived("name", "Nicola");
+ compensate.expectedMessagesMatches(ex -> ex.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION) != null);
+
+ try {
+ template.sendBodyAndHeader("direct:workflow", "compensate", "myname", "Nicola");
+ Assert.fail("Should throw an exception");
+ } catch (Exception ex) {
+ // OK
+ }
+
+ compensate.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ context.addService(new InMemorySagaService());
+
+ from("direct:workflow")
+ .saga()
+ .option("id", constant("myheader"))
+ .option("name", header("myname"))
+ .completion("mock:complete")
+ .compensation("mock:compensate")
+ .choice()
+ .when(body().isEqualTo("compensate"))
+ .process(ex -> {
+ throw new RuntimeException("forced compensate");
+ })
+ .end()
+ .setHeader("myname", constant("TryToOverride"))
+ .setHeader("name", constant("TryToOverride"))
+ .to("mock:endpoint");
+
+ }
+ };
+ }
+
+}
diff --git a/camel-core/src/test/java/org/apache/camel/processor/SagaPropagationTest.java b/camel-core/src/test/java/org/apache/camel/processor/SagaPropagationTest.java
new file mode 100644
index 0000000..f91733f
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/SagaPropagationTest.java
@@ -0,0 +1,197 @@
+/**
+ * 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 java.util.LinkedList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.saga.InMemorySagaService;
+import org.apache.camel.model.SagaPropagation;
+import org.apache.camel.saga.CamelSagaService;
+import org.junit.Assert;
+
+public class SagaPropagationTest extends ContextTestSupport {
+
+ private List<String> sagaIds;
+
+
+ public void testPropagationRequired() throws Exception {
+ context.createFluentProducerTemplate().to("direct:required").request();
+
+ assertListSize(sagaIds, 3);
+ assertUniqueNonNullSagaIds(1);
+ }
+
+ public void testPropagationRequiresNew() throws Exception {
+ context.createFluentProducerTemplate().to("direct:requiresNew").request();
+
+ assertListSize(sagaIds, 3);
+ assertUniqueNonNullSagaIds(3);
+ }
+
+ public void testPropagationNotSupported() throws Exception {
+ context.createFluentProducerTemplate().to("direct:notSupported").request();
+
+ assertListSize(sagaIds, 4);
+ assertNonNullSagaIds(1);
+ }
+
+ public void testPropagationSupports() throws Exception {
+ context.createFluentProducerTemplate().to("direct:supports").request();
+
+ assertListSize(sagaIds, 2);
+ assertNonNullSagaIds(1);
+ }
+
+ public void testPropagationMandatory() throws Exception {
+ try {
+ context.createFluentProducerTemplate().to("direct:mandatory").request();
+ Assert.fail("Exception not thrown");
+ } catch (CamelExecutionException ex) {
+ // fine
+ }
+ }
+
+ public void testPropagationNever() throws Exception {
+ try {
+ context.createFluentProducerTemplate().to("direct:never").request();
+ Assert.fail("Exception not thrown");
+ } catch (CamelExecutionException ex) {
+ // fine
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ this.sagaIds = new LinkedList<>();
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ CamelSagaService sagaService = new InMemorySagaService();
+ context.addService(sagaService);
+
+ // REQUIRED
+
+ from("direct:required")
+ .saga()
+ .process(addSagaIdToList())
+ .to("direct:required2");
+
+ from("direct:required2")
+ .saga().propagation(SagaPropagation.REQUIRED)
+ .process(addSagaIdToList())
+ .to("direct:required3");
+
+ from("direct:required3")
+ .saga()
+ .process(addSagaIdToList());
+
+ // REQUIRES_NEW
+
+ from("direct:requiresNew")
+ .saga().propagation(SagaPropagation.REQUIRES_NEW)
+ .process(addSagaIdToList())
+ .to("direct:requiresNew2")
+ .to("direct:requiresNew2");
+
+ from("direct:requiresNew2")
+ .saga().propagation(SagaPropagation.REQUIRES_NEW)
+ .process(addSagaIdToList());
+
+ // NOT_SUPPORTED
+
+ from("direct:notSupported")
+ .process(addSagaIdToList())
+ .to("direct:notSupported2")
+ .to("direct:notSupported3");
+
+ from("direct:notSupported2")
+ .saga() // required
+ .process(addSagaIdToList())
+ .to("direct:notSupported3");
+
+ from("direct:notSupported3")
+ .saga().propagation(SagaPropagation.NOT_SUPPORTED)
+ .process(addSagaIdToList());
+
+ // SUPPORTS
+
+ from("direct:supports")
+ .to("direct:supports2")
+ .to("direct:supports3");
+
+ from("direct:supports2")
+ .saga() // required
+ .to("direct:supports3");
+
+ from("direct:supports3")
+ .saga().propagation(SagaPropagation.SUPPORTS)
+ .process(addSagaIdToList());
+
+ // MANDATORY
+
+ from("direct:mandatory")
+ .to("direct:mandatory2");
+
+ from("direct:mandatory2")
+ .saga().propagation(SagaPropagation.MANDATORY)
+ .process(addSagaIdToList());
+
+ // NEVER
+
+ from("direct:never")
+ .saga()
+ .to("direct:never2");
+
+ from("direct:never2")
+ .saga().propagation(SagaPropagation.NEVER)
+ .process(addSagaIdToList());
+
+ }
+ };
+ }
+
+ private Processor addSagaIdToList() {
+ return ex -> sagaIds.add(ex.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION, String.class));
+ }
+
+ private void assertUniqueNonNullSagaIds(int num) {
+ Set<String> uniqueNonNull = this.sagaIds.stream().filter(Objects::nonNull).collect(Collectors.toSet());
+ if (uniqueNonNull.size() != num) {
+ Assert.fail("Expeced size " + num + ", actual " + uniqueNonNull.size());
+ }
+ }
+
+ private void assertNonNullSagaIds(int num) {
+ List<String> nonNull = this.sagaIds.stream().filter(Objects::nonNull).collect(Collectors.toList());
+ if (nonNull.size() != num) {
+ Assert.fail("Expeced size " + num + ", actual " + nonNull.size());
+ }
+ }
+
+}
diff --git a/camel-core/src/test/java/org/apache/camel/processor/SagaTest.java b/camel-core/src/test/java/org/apache/camel/processor/SagaTest.java
new file mode 100644
index 0000000..10415ee
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/SagaTest.java
@@ -0,0 +1,200 @@
+/**
+ * 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 java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Header;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.saga.InMemorySagaService;
+import org.apache.camel.model.SagaPropagation;
+import org.apache.camel.saga.CamelSagaService;
+import org.junit.Assert;
+
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.Matchers.equalTo;
+
+public class SagaTest extends ContextTestSupport {
+
+ private OrderManagerService orderManagerService;
+
+ private CreditService creditService;
+
+ public void testCreditExhausted() throws Exception {
+ // total credit is 100
+ buy(20, false, false);
+ buy(70, false, false);
+ buy(20, false, true); // fail
+ buy(5, false, false);
+
+ await().until(() -> orderManagerService.getOrders().size(), equalTo(3));
+ await().until(() -> creditService.getCredit(), equalTo(5));
+ }
+
+ public void testTotalCompensation() throws Exception {
+ // total credit is 100
+ for (int i = 0; i < 10; i++) {
+ if (i % 2 == 0) {
+ buy(10, false, false);
+ } else {
+ buy(10, true, true);
+ }
+ }
+
+ await().until(() -> orderManagerService.getOrders().size(), equalTo(5));
+ await().until(() -> creditService.getCredit(), equalTo(50));
+ }
+
+ private void buy(int amount, boolean failAtTheEnd, boolean shouldFail) {
+ try {
+ context.createFluentProducerTemplate()
+ .to("direct:saga")
+ .withHeader("amount", amount)
+ .withHeader("fail", failAtTheEnd)
+ .request();
+
+ if (shouldFail) {
+ Assert.fail("Exception not thrown");
+ }
+ } catch (Exception ex) {
+ if (!shouldFail) {
+ Assert.fail("Unexpected exception");
+ }
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ orderManagerService = new OrderManagerService();
+
+ creditService = new CreditService(100);
+
+ CamelSagaService sagaService = new InMemorySagaService();
+ context.addService(sagaService);
+
+ from("direct:saga")
+ .saga().propagation(SagaPropagation.REQUIRES_NEW)
+ .log("Creating a new order")
+ .to("direct:newOrder")
+ .log("Taking the credit")
+ .to("direct:reserveCredit")
+ .log("Finalizing")
+ .to("direct:finalize")
+ .log("Done!");
+
+
+ // Order service
+
+ from("direct:newOrder")
+ .saga()
+ .propagation(SagaPropagation.MANDATORY)
+ .compensation("direct:cancelOrder")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(orderManagerService, "newOrder")
+ .log("Order ${body} created");
+
+ from("direct:cancelOrder")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(orderManagerService, "cancelOrder")
+ .log("Order ${body} cancelled");
+
+
+ // Credit service
+
+ from("direct:reserveCredit")
+ .saga()
+ .propagation(SagaPropagation.MANDATORY)
+ .compensation("direct:refundCredit")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(creditService, "reserveCredit")
+ .log("Credit ${header.amount} reserved in action ${body}");
+
+ from("direct:refundCredit")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(creditService, "refundCredit")
+ .log("Credit for action ${body} refunded");
+
+
+ // Final actions
+ from("direct:finalize")
+ .saga().propagation(SagaPropagation.NOT_SUPPORTED)
+ .choice()
+ .when(header("fail").isEqualTo(true))
+ .process(x -> {
+ throw new RuntimeException("fail");
+ })
+ .end();
+
+ }
+ };
+ }
+
+ public static class OrderManagerService {
+
+ private Set<String> orders = new HashSet<>();
+
+ public synchronized void newOrder(String id) {
+ orders.add(id);
+ }
+
+ public synchronized void cancelOrder(String id) {
+ orders.remove(id);
+ }
+
+ public synchronized Set<String> getOrders() {
+ return new TreeSet<>(orders);
+ }
+ }
+
+ public static class CreditService {
+
+ private int totalCredit;
+
+ private Map<String, Integer> reservations = new HashMap<>();
+
+ public CreditService(int totalCredit) {
+ this.totalCredit = totalCredit;
+ }
+
+ public synchronized void reserveCredit(String id, @Header("amount") int amount) {
+ int credit = getCredit();
+ if (amount > credit) {
+ throw new IllegalStateException("Insufficient credit");
+ }
+ reservations.put(id, amount);
+ }
+
+ public synchronized void refundCredit(String id) {
+ reservations.remove(id);
+ }
+
+ public synchronized int getCredit() {
+ return totalCredit - reservations.values().stream().reduce(0, (a, b) -> a + b);
+ }
+ }
+}
diff --git a/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java b/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java
new file mode 100644
index 0000000..936c4e0
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/SagaTimeoutTest.java
@@ -0,0 +1,94 @@
+/**
+ * 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 java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.saga.InMemorySagaService;
+import org.apache.camel.model.SagaCompletionMode;
+
+public class SagaTimeoutTest extends ContextTestSupport {
+
+
+ public void testTimeoutCalledCorrectly() throws Exception {
+
+ MockEndpoint compensate = getMockEndpoint("mock:compensate");
+ compensate.expectedMessageCount(1);
+ compensate.expectedHeaderReceived("id", "myid");
+
+ MockEndpoint end = getMockEndpoint("mock:end");
+ end.expectedMessageCount(1);
+
+ template.sendBody("direct:saga", "Hello");
+
+ end.assertIsSatisfied();
+ compensate.assertIsSatisfied();
+ }
+
+ public void testTimeoutHasNoEffectIfCompleted() throws Exception {
+
+ MockEndpoint compensate = getMockEndpoint("mock:compensate");
+ compensate.expectedMessageCount(1);
+ compensate.setResultWaitTime(500);
+
+ MockEndpoint complete = getMockEndpoint("mock:complete");
+ complete.expectedMessageCount(1);
+ complete.expectedHeaderReceived("id", "myid");
+
+ MockEndpoint end = getMockEndpoint("mock:end");
+ end.expectedMessageCount(1);
+
+ template.sendBody("direct:saga-auto", "Hello");
+
+ end.assertIsSatisfied();
+ complete.assertIsSatisfied();
+ compensate.assertIsNotSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ context.addService(new InMemorySagaService());
+
+ from("direct:saga")
+ .saga()
+ .timeout(100, TimeUnit.MILLISECONDS)
+ .option("id", constant("myid"))
+ .completionMode(SagaCompletionMode.MANUAL)
+ .compensation("mock:compensate")
+ .to("mock:end");
+
+ from("direct:saga-auto")
+ .saga()
+ .timeout(350, TimeUnit.MILLISECONDS)
+ .option("id", constant("myid"))
+ .compensation("mock:compensate")
+ .completion("mock:complete")
+ .to("mock:end");
+
+ }
+ };
+ }
+
+}
diff --git a/camel-core/src/test/resources/org/apache/camel/model/saga.xml b/camel-core/src/test/resources/org/apache/camel/model/saga.xml
new file mode 100644
index 0000000..03412a8
--- /dev/null
+++ b/camel-core/src/test/resources/org/apache/camel/model/saga.xml
@@ -0,0 +1,31 @@
+<?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.
+
+-->
+<routes id="camel" xmlns="http://camel.apache.org/schema/spring">
+ <route>
+ <from uri="direct:start"/>
+ <saga>
+ <compensation uri="mock:uri" />
+ </saga>
+ <setBody>
+ <simple>${in.body} extra data!</simple>
+ </setBody>
+ <to uri="mock:end"/>
+ </route>
+</routes>
diff --git a/components/camel-lra/pom.xml b/components/camel-lra/pom.xml
new file mode 100644
index 0000000..a536d3a
--- /dev/null
+++ b/components/camel-lra/pom.xml
@@ -0,0 +1,122 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>2.21.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-lra</artifactId>
+ <packaging>jar</packaging>
+ <name>Camel :: Long-Running-Action</name>
+ <description>Camel saga binding for Long-Running-Action framework</description>
+
+ <properties>
+ <!-- used by camel-catalog -->
+ <firstVersion>2.21.0</firstVersion>
+
+ <camel.osgi.export.pkg>org.apache.camel.service.lra.*</camel.osgi.export.pkg>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.cxf</groupId>
+ <artifactId>cxf-rt-rs-client</artifactId>
+ </dependency>
+
+
+ <!-- test dependencies -->
+
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-undertow</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-jackson</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <profiles>
+ <profile>
+ <id>lra-integration-tests</id>
+ <activation>
+ <property>
+ <name>env.LRA_COORDINATOR_URL</name>
+ </property>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <includes>
+ <include>**/*Test.java</include>
+ <include>**/*IT.java</include>
+ </includes>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+
+</project>
\ No newline at end of file
diff --git a/components/camel-lra/src/main/docs/lra.adoc b/components/camel-lra/src/main/docs/lra.adoc
new file mode 100644
index 0000000..892ac18
--- /dev/null
+++ b/components/camel-lra/src/main/docs/lra.adoc
@@ -0,0 +1,18 @@
+## LRA Component
+
+*Available as of Camel version 2.21.0*
+
+The LRA module provides bindings of the Saga EIP with any https://github.com/eclipse/microprofile-sandbox/tree/master/proposals/0009-LRA[Microprofile compatible LRA Coordinator (sandbox)].
+
+Maven users will need to add the following dependency to their `pom.xml`
+for this component:
+
+[source,xml]
+------------------------------------------------------------
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-lra</artifactId>
+ <version>x.x.x</version>
+ <!-- use the same version as your Camel core version -->
+</dependency>
+------------------------------------------------------------
\ No newline at end of file
diff --git a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAClient.java b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAClient.java
new file mode 100644
index 0000000..7c2edf5
--- /dev/null
+++ b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAClient.java
@@ -0,0 +1,191 @@
+/**
+ * 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.service.lra;
+
+import java.net.URL;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.client.InvocationCallback;
+import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.RuntimeCamelException;
+
+import static org.apache.camel.service.lra.LRAConstants.COORDINATOR_PATH_CANCEL;
+import static org.apache.camel.service.lra.LRAConstants.COORDINATOR_PATH_CLOSE;
+import static org.apache.camel.service.lra.LRAConstants.COORDINATOR_PATH_START;
+import static org.apache.camel.service.lra.LRAConstants.HEADER_LINK;
+import static org.apache.camel.service.lra.LRAConstants.HEADER_TIME_LIMIT;
+import static org.apache.camel.service.lra.LRAConstants.PARTICIPANT_PATH_COMPENSATE;
+import static org.apache.camel.service.lra.LRAConstants.PARTICIPANT_PATH_COMPLETE;
+
+/**
+ *
+ */
+public class LRAClient {
+
+
+ private final LRASagaService sagaService;
+
+ private final Client client;
+
+ private final WebTarget target;
+
+ public LRAClient(LRASagaService sagaService) {
+ this.sagaService = sagaService;
+
+ this.client = ClientBuilder.newBuilder()
+ .executorService(sagaService.getExecutorService())
+ .build();
+
+ this.target = client.target(
+ new LRAUrlBuilder()
+ .host(sagaService.getCoordinatorUrl())
+ .path(sagaService.getCoordinatorContextPath())
+ .build()
+ );
+ }
+
+ public CompletableFuture<URL> newLRA() {
+ CompletableFuture<Response> future = new CompletableFuture<>();
+ target.path(COORDINATOR_PATH_START)
+ .request()
+ .async()
+ .post(Entity.text(""), callbackToCompletableFuture(future));
+
+ return future.thenApply(res -> {
+ URL lraURL = toURL(res.getHeaders().getFirst(Exchange.SAGA_LONG_RUNNING_ACTION));
+ if (lraURL == null) {
+ throw new IllegalStateException("Cannot obtain LRA id from LRA coordinator");
+ }
+
+ return lraURL;
+ });
+ }
+
+ public CompletableFuture<Void> join(URL lra, LRASagaStep step) {
+ return CompletableFuture.supplyAsync(() -> {
+ LRAUrlBuilder participantBaseUrl = new LRAUrlBuilder()
+ .host(sagaService.getLocalParticipantUrl())
+ .path(sagaService.getLocalParticipantContextPath())
+ .options(step.getOptions())
+ .compensation(step.getCompensation())
+ .completion(step.getCompletion());
+
+
+ String compensationURL = participantBaseUrl.path(PARTICIPANT_PATH_COMPENSATE).build();
+ String completionURL = participantBaseUrl.path(PARTICIPANT_PATH_COMPLETE).build();
+
+ StringBuilder link = new StringBuilder();
+ link.append('<').append(compensationURL).append('>').append("; rel=compensate");
+ link.append(',');
+ link.append('<').append(completionURL).append('>').append("; rel=complete");
+
+
+ WebTarget joinTarget = client.target(lra.toString());
+ if (step.getTimeoutInMilliseconds().isPresent()) {
+ joinTarget = joinTarget.queryParam(HEADER_TIME_LIMIT, step.getTimeoutInMilliseconds().get());
+ }
+
+ CompletableFuture<Response> future = new CompletableFuture<>();
+ joinTarget.request()
+ .header(HEADER_LINK, link.toString())
+ .header(Exchange.SAGA_LONG_RUNNING_ACTION, lra)
+ .async()
+ .put(Entity.entity(link.toString(), MediaType.TEXT_PLAIN), callbackToCompletableFuture(future));
+
+ return future;
+ }, sagaService.getExecutorService())
+ .thenCompose(Function.identity())
+ .thenApply(response -> {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
+ throw new RuntimeCamelException("Cannot join LRA");
+ }
+
+ return null;
+ });
+ }
+
+ public CompletableFuture<Void> complete(URL lra) {
+ CompletableFuture<Response> future = new CompletableFuture<>();
+ client.target(lra.toString())
+ .path(COORDINATOR_PATH_CLOSE)
+ .request()
+ .async()
+ .put(Entity.entity("", MediaType.TEXT_PLAIN), callbackToCompletableFuture(future));
+
+ return future.thenApply(response -> {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
+ throw new RuntimeCamelException("Cannot complete LRA");
+ }
+
+ return null;
+ });
+ }
+
+ public CompletableFuture<Void> compensate(URL lra) {
+ CompletableFuture<Response> future = new CompletableFuture<>();
+ client.target(lra.toString())
+ .path(COORDINATOR_PATH_CANCEL)
+ .request()
+ .async()
+ .put(Entity.entity("", MediaType.TEXT_PLAIN), callbackToCompletableFuture(future));
+
+ return future.thenApply(response -> {
+ if (response.getStatus() != Response.Status.OK.getStatusCode()) {
+ throw new RuntimeCamelException("Cannot compensate LRA");
+ }
+
+ return null;
+ });
+ }
+
+ private InvocationCallback<Response> callbackToCompletableFuture(CompletableFuture<Response> future) {
+ return new InvocationCallback<Response>() {
+ @Override
+ public void completed(Response response) {
+ future.complete(response);
+ }
+
+ @Override
+ public void failed(Throwable throwable) {
+ future.completeExceptionally(throwable);
+ }
+ };
+ }
+
+ private URL toURL(Object url) {
+ if (url == null) {
+ return null;
+ }
+ if (url instanceof URL) {
+ return URL.class.cast(url);
+ }
+
+ try {
+ return new URL(url.toString());
+ } catch (Exception ex) {
+ throw new RuntimeCamelException(ex);
+ }
+ }
+
+}
diff --git a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAConstants.java b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAConstants.java
new file mode 100644
index 0000000..748689c
--- /dev/null
+++ b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAConstants.java
@@ -0,0 +1,43 @@
+/**
+ * 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.service.lra;
+
+/**
+ *
+ */
+public final class LRAConstants {
+
+ public static final String DEFAULT_COORDINATOR_CONTEXT_PATH = "/lra-coordinator";
+ public static final String DEFAULT_LOCAL_PARTICIPANT_CONTEXT_PATH = "/lra-participant";
+
+ static final String COORDINATOR_PATH_START = "/start";
+ static final String COORDINATOR_PATH_CLOSE = "/close";
+ static final String COORDINATOR_PATH_CANCEL = "/cancel";
+
+ static final String PARTICIPANT_PATH_COMPENSATE = "/compensate";
+ static final String PARTICIPANT_PATH_COMPLETE = "/complete";
+
+ static final String HEADER_LINK = "Link";
+ static final String HEADER_TIME_LIMIT = "TimeLimit";
+
+ static final String URL_COMPENSATION_KEY = "Camel-Saga-Compensate";
+ static final String URL_COMPLETION_KEY = "Camel-Saga-Complete";
+
+ private LRAConstants() {
+ }
+
+}
diff --git a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaCoordinator.java b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaCoordinator.java
new file mode 100644
index 0000000..1442454
--- /dev/null
+++ b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaCoordinator.java
@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.service.lra;
+
+import java.net.URL;
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.saga.CamelSagaCoordinator;
+import org.apache.camel.saga.CamelSagaStep;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ *
+ */
+public class LRASagaCoordinator implements CamelSagaCoordinator {
+
+ private LRASagaService sagaService;
+
+ private URL lraURL;
+
+ public LRASagaCoordinator(LRASagaService sagaService, URL lraURL) {
+ this.sagaService = ObjectHelper.notNull(sagaService, "sagaService");
+ this.lraURL = ObjectHelper.notNull(lraURL, "lraURL");
+ }
+
+ @Override
+ public CompletableFuture<Void> beginStep(Exchange exchange, CamelSagaStep step) {
+ LRASagaStep sagaStep = LRASagaStep.fromCamelSagaStep(step, exchange);
+ return sagaService.getClient().join(this.lraURL, sagaStep);
+ }
+
+ @Override
+ public CompletableFuture<Void> compensate() {
+ return sagaService.getClient().compensate(this.lraURL);
+ }
+
+ @Override
+ public CompletableFuture<Void> complete() {
+ return sagaService.getClient().complete(this.lraURL);
+ }
+
+ @Override
+ public String getId() {
+ return this.lraURL.toString();
+ }
+}
diff --git a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java
new file mode 100644
index 0000000..182e8ed
--- /dev/null
+++ b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaRoutes.java
@@ -0,0 +1,89 @@
+/**
+ * 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.service.lra;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+
+import static org.apache.camel.service.lra.LRAConstants.PARTICIPANT_PATH_COMPENSATE;
+import static org.apache.camel.service.lra.LRAConstants.PARTICIPANT_PATH_COMPLETE;
+import static org.apache.camel.service.lra.LRAConstants.URL_COMPENSATION_KEY;
+import static org.apache.camel.service.lra.LRAConstants.URL_COMPLETION_KEY;
+
+/**
+ *
+ */
+public class LRASagaRoutes extends RouteBuilder {
+
+ private LRASagaService sagaService;
+
+ public LRASagaRoutes(LRASagaService sagaService) {
+ this.sagaService = sagaService;
+ }
+
+ @Override
+ public void configure() throws Exception {
+
+ rest(sagaService.getLocalParticipantContextPath())
+ .put(PARTICIPANT_PATH_COMPENSATE)
+ .route().id("lra-compensation")
+ .process(this::verifyRequest)
+ .choice()
+ .when(header(URL_COMPENSATION_KEY).isNotNull())
+ .toD("header." + URL_COMPENSATION_KEY)
+ .end();
+
+ rest(sagaService.getLocalParticipantContextPath())
+ .put(PARTICIPANT_PATH_COMPLETE)
+ .route().id("lra-completion")
+ .process(this::verifyRequest)
+ .choice()
+ .when(header(URL_COMPLETION_KEY).isNotNull())
+ .toD("header." + URL_COMPLETION_KEY)
+ .end();
+
+ }
+
+ /**
+ * Check if the request is pointing to an allowed URI to prevent unauthorized remote uri invocation
+ */
+ private void verifyRequest(Exchange exchange) {
+ if (exchange.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION) == null) {
+ throw new IllegalArgumentException("Missing " + Exchange.SAGA_LONG_RUNNING_ACTION + " header in received request");
+ }
+
+ Set<String> usedURIs = new HashSet<>();
+ String compensationURI = exchange.getIn().getHeader(URL_COMPENSATION_KEY, String.class);
+ if (compensationURI != null) {
+ usedURIs.add(compensationURI);
+ }
+ String completionURI = exchange.getIn().getHeader(URL_COMPLETION_KEY, String.class);
+ if (completionURI != null) {
+ usedURIs.add(completionURI);
+ }
+
+ for (String uri : usedURIs) {
+ if (!sagaService.getRegisteredURIs().contains(uri)) {
+ throw new IllegalArgumentException("URI " + uri + " is not allowed");
+ }
+ }
+ }
+
+}
diff --git a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java
new file mode 100644
index 0000000..28a33b5
--- /dev/null
+++ b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaService.java
@@ -0,0 +1,168 @@
+/**
+ * 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.service.lra;
+
+import java.net.URL;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.saga.CamelSagaCoordinator;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.saga.CamelSagaStep;
+import org.apache.camel.support.ServiceSupport;
+
+/**
+ * A Camel saga service based on LRA (https://github.com/eclipse/microprofile-sandbox/tree/master/proposals/0009-LRA).
+ */
+public class LRASagaService extends ServiceSupport implements CamelSagaService {
+
+ private CamelContext camelContext;
+
+ private ScheduledExecutorService executorService;
+
+ private LRAClient client;
+
+ private LRASagaRoutes routes;
+
+ private String coordinatorUrl;
+
+ private String coordinatorContextPath = LRAConstants.DEFAULT_COORDINATOR_CONTEXT_PATH;
+
+ private String localParticipantUrl;
+
+ private String localParticipantContextPath = LRAConstants.DEFAULT_LOCAL_PARTICIPANT_CONTEXT_PATH;
+
+ private Set<String> sagaURIs = ConcurrentHashMap.newKeySet();
+
+
+ public LRASagaService() {
+ }
+
+ @Override
+ public CompletableFuture<CamelSagaCoordinator> newSaga() {
+ return client.newLRA()
+ .thenApply(url -> new LRASagaCoordinator(LRASagaService.this, url));
+ }
+
+ @Override
+ public CompletableFuture<CamelSagaCoordinator> getSaga(String id) {
+ CompletableFuture<CamelSagaCoordinator> coordinator = new CompletableFuture<>();
+ try {
+ coordinator.complete(new LRASagaCoordinator(this, new URL(id)));
+ } catch (Exception ex) {
+ coordinator.completeExceptionally(ex);
+ }
+ return coordinator;
+ }
+
+ @Override
+ public void registerStep(CamelSagaStep step) {
+ // Register which uris should be exposed
+ step.getCompensation().map(Endpoint::getEndpointUri).map(this.sagaURIs::add);
+ step.getCompletion().map(Endpoint::getEndpointUri).map(this.sagaURIs::add);
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ if (this.executorService == null) {
+ this.executorService = camelContext.getExecutorServiceManager()
+ .newDefaultScheduledThreadPool(this, "saga-lra");
+ }
+ if (this.client == null) {
+ this.client = new LRAClient(this);
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (this.executorService != null) {
+ camelContext.getExecutorServiceManager().shutdownGraceful(this.executorService);
+ this.executorService = null;
+ }
+ if (this.client != null) {
+ this.client = null;
+ }
+ }
+
+ @Override
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ if (this.routes == null) {
+ this.routes = new LRASagaRoutes(this);
+ try {
+ this.camelContext.addRoutes(this.routes);
+ } catch (Exception ex) {
+ throw new RuntimeCamelException(ex);
+ }
+ }
+ }
+
+ @Override
+ public CamelContext getCamelContext() {
+ return this.camelContext;
+ }
+
+ public ScheduledExecutorService getExecutorService() {
+ return executorService;
+ }
+
+ public LRAClient getClient() {
+ return client;
+ }
+
+ public String getCoordinatorUrl() {
+ return coordinatorUrl;
+ }
+
+ public void setCoordinatorUrl(String coordinatorUrl) {
+ this.coordinatorUrl = coordinatorUrl;
+ }
+
+ public String getCoordinatorContextPath() {
+ return coordinatorContextPath;
+ }
+
+ public void setCoordinatorContextPath(String coordinatorContextPath) {
+ this.coordinatorContextPath = coordinatorContextPath;
+ }
+
+ public String getLocalParticipantUrl() {
+ return localParticipantUrl;
+ }
+
+ public void setLocalParticipantUrl(String localParticipantUrl) {
+ this.localParticipantUrl = localParticipantUrl;
+ }
+
+ public String getLocalParticipantContextPath() {
+ return localParticipantContextPath;
+ }
+
+ public void setLocalParticipantContextPath(String localParticipantContextPath) {
+ this.localParticipantContextPath = localParticipantContextPath;
+ }
+
+ public Set<String> getRegisteredURIs() {
+ return sagaURIs;
+ }
+
+}
diff --git a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaStep.java b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaStep.java
new file mode 100644
index 0000000..e4b182d
--- /dev/null
+++ b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRASagaStep.java
@@ -0,0 +1,81 @@
+/**
+ * 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.service.lra;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.TreeMap;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.saga.CamelSagaStep;
+
+/**
+ *
+ */
+public final class LRASagaStep {
+
+ private Optional<Endpoint> compensation;
+
+ private Optional<Endpoint> completion;
+
+ private Map<String, String> options;
+
+ private Optional<Long> timeoutInMilliseconds;
+
+ private LRASagaStep() {
+ }
+
+ public static LRASagaStep fromCamelSagaStep(CamelSagaStep step, Exchange exchange) {
+ LRASagaStep t = new LRASagaStep();
+ t.compensation = step.getCompensation();
+ t.completion = step.getCompletion();
+ t.timeoutInMilliseconds = step.getTimeoutInMilliseconds();
+ t.options = new TreeMap<>();
+ for (Map.Entry<String, Expression> entry : step.getOptions().entrySet()) {
+ t.options.put(entry.getKey(), entry.getValue().evaluate(exchange, String.class));
+ }
+ return t;
+ }
+
+ public Optional<Endpoint> getCompensation() {
+ return compensation;
+ }
+
+ public Optional<Endpoint> getCompletion() {
+ return completion;
+ }
+
+ public Map<String, String> getOptions() {
+ return options;
+ }
+
+ public Optional<Long> getTimeoutInMilliseconds() {
+ return timeoutInMilliseconds;
+ }
+
+ @Override
+ public String toString() {
+ return "LRASagaStep{"
+ + "compensation=" + compensation
+ + ", completion=" + completion
+ + ", options=" + options
+ + ", timeoutInMilliseconds=" + timeoutInMilliseconds
+ + '}';
+ }
+}
diff --git a/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAUrlBuilder.java b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAUrlBuilder.java
new file mode 100644
index 0000000..156f825
--- /dev/null
+++ b/components/camel-lra/src/main/java/org/apache/camel/service/lra/LRAUrlBuilder.java
@@ -0,0 +1,139 @@
+/**
+ * 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.service.lra;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.Optional;
+
+import org.apache.camel.Endpoint;
+
+import static org.apache.camel.service.lra.LRAConstants.URL_COMPENSATION_KEY;
+import static org.apache.camel.service.lra.LRAConstants.URL_COMPLETION_KEY;
+
+/**
+ *
+ */
+public class LRAUrlBuilder {
+
+ private String host;
+
+ private String path = "";
+
+ private String query = "";
+
+ public LRAUrlBuilder() {
+ }
+
+ public LRAUrlBuilder(String host, String path, String query) {
+ this.host = host;
+ this.path = path;
+ this.query = query;
+ }
+
+ public LRAUrlBuilder host(String host) {
+ if (this.host != null) {
+ throw new IllegalStateException("Host already set");
+ }
+ LRAUrlBuilder copy = copy();
+ copy.host = host;
+ return copy;
+ }
+
+ public LRAUrlBuilder path(String path) {
+ LRAUrlBuilder copy = copy();
+ copy.path = joinPath(this.path, path);
+ return copy;
+ }
+
+ public LRAUrlBuilder compensation(Optional<Endpoint> endpoint) {
+ if (endpoint.isPresent()) {
+ return compensation(endpoint.get().getEndpointUri());
+ }
+ return this;
+ }
+
+ public LRAUrlBuilder compensation(String uri) {
+ return this.query(URL_COMPENSATION_KEY, uri);
+ }
+
+ public LRAUrlBuilder completion(Optional<Endpoint> endpoint) {
+ if (endpoint.isPresent()) {
+ return completion(endpoint.get().getEndpointUri());
+ }
+ return this;
+ }
+
+ public LRAUrlBuilder completion(String uri) {
+ return this.query(URL_COMPLETION_KEY, uri);
+ }
+
+ public LRAUrlBuilder options(Map<String, ?> options) {
+ LRAUrlBuilder result = this;
+ for (Map.Entry<String, ?> entry : options.entrySet()) {
+ result = result.query(entry.getKey(), entry.getValue());
+ }
+ return result;
+ }
+
+ public LRAUrlBuilder query(String key, Object value) {
+ LRAUrlBuilder copy = copy();
+ try {
+ key = URLEncoder.encode(toNonnullString(key), StandardCharsets.UTF_8.name());
+ value = URLEncoder.encode(toNonnullString(value), StandardCharsets.UTF_8.name());
+ if (copy.query.length() == 0) {
+ copy.query += "?";
+ } else {
+ copy.query += "&";
+ }
+ copy.query += key + "=" + value;
+ } catch (UnsupportedEncodingException e) {
+ throw new IllegalStateException(e);
+ }
+ return copy;
+ }
+
+ public String build() {
+ if (this.host == null) {
+ throw new IllegalStateException("Host not set");
+ }
+ return joinPath(this.host, this.path) + query;
+ }
+
+ private String joinPath(String first, String second) {
+ first = toNonnullString(first);
+ second = toNonnullString(second);
+ while (first.endsWith("/")) {
+ first = first.substring(0, first.length() - 1);
+ }
+ while (second.startsWith("/")) {
+ second = second.substring(1);
+ }
+ return first + "/" + second;
+ }
+
+ private String toNonnullString(Object obj) {
+ return obj != null ? obj.toString() : "";
+ }
+
+ private LRAUrlBuilder copy() {
+ return new LRAUrlBuilder(this.host, this.path, this.query);
+ }
+
+}
diff --git a/components/camel-lra/src/main/resources/META-INF/LICENSE.txt b/components/camel-lra/src/main/resources/META-INF/LICENSE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/components/camel-lra/src/main/resources/META-INF/LICENSE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed 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.
+
diff --git a/components/camel-lra/src/main/resources/META-INF/NOTICE.txt b/components/camel-lra/src/main/resources/META-INF/NOTICE.txt
new file mode 100644
index 0000000..2e215bf
--- /dev/null
+++ b/components/camel-lra/src/main/resources/META-INF/NOTICE.txt
@@ -0,0 +1,11 @@
+ =========================================================================
+ == NOTICE file corresponding to the section 4 d of ==
+ == the Apache License, Version 2.0, ==
+ == in this case for the Apache Camel distribution. ==
+ =========================================================================
+
+ This product includes software developed by
+ The Apache Software Foundation (http://www.apache.org/).
+
+ Please read the different LICENSE files present in the licenses directory of
+ this distribution.
diff --git a/components/camel-lra/src/test/java/org/apache/camel/service/lra/AbstractLRATestSupport.java b/components/camel-lra/src/test/java/org/apache/camel/service/lra/AbstractLRATestSupport.java
new file mode 100644
index 0000000..287b467
--- /dev/null
+++ b/components/camel-lra/src/test/java/org/apache/camel/service/lra/AbstractLRATestSupport.java
@@ -0,0 +1,105 @@
+/**
+ * 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.service.lra;
+
+import java.io.IOException;
+import java.io.InputStream;
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.core.Response;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+
+/**
+ * Base class for LRA based tests.
+ */
+public abstract class AbstractLRATestSupport extends CamelTestSupport {
+
+ private Integer serverPort;
+
+ private int activeLRAs;
+
+ @Before
+ public void getActiveLRAs() throws IOException {
+ this.activeLRAs = getNumberOfActiveLRAs();
+ }
+
+ @After
+ public void checkActiveLRAs() throws IOException {
+ Assert.assertEquals("Some LRA have been left pending", activeLRAs, getNumberOfActiveLRAs());
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+
+ context.addService(createLRASagaService());
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ restConfiguration()
+ .port(getServerPort());
+ }
+ });
+
+ return context;
+ }
+
+ protected LRASagaService createLRASagaService() {
+ LRASagaService sagaService = new LRASagaService();
+ sagaService.setCoordinatorUrl(getCoordinatorURL());
+ sagaService.setLocalParticipantUrl("http://localhost:" + getServerPort());
+ return sagaService;
+ }
+
+ protected int getNumberOfActiveLRAs() throws IOException {
+ Client client = ClientBuilder.newClient();
+
+ Response response = client.target(getCoordinatorURL() + "/lra-coordinator")
+ .request()
+ .accept("application/json")
+ .get();
+
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode lras = mapper.readTree(InputStream.class.cast(response.getEntity()));
+ return lras.size();
+ }
+
+ private String getCoordinatorURL() {
+ String url = System.getenv("LRA_COORDINATOR_URL");
+ if (url == null) {
+ throw new IllegalStateException("Cannot run test: environment variable LRA_COORDINATOR_URL is missing");
+ }
+ return url;
+ }
+
+ protected int getServerPort() {
+ if (serverPort == null) {
+ serverPort = AvailablePortFinder.getNextAvailable();
+ }
+ return serverPort;
+ }
+}
diff --git a/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRACreditIT.java b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRACreditIT.java
new file mode 100644
index 0000000..b0d4254
--- /dev/null
+++ b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRACreditIT.java
@@ -0,0 +1,196 @@
+/**
+ * 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.service.lra;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Header;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.SagaPropagation;
+import org.junit.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.Matchers.equalTo;
+
+public class LRACreditIT extends AbstractLRATestSupport {
+
+ private OrderManagerService orderManagerService;
+
+ private CreditService creditService;
+
+ @Test
+ public void testCreditExhausted() throws Exception {
+ // total credit is 100
+ buy(20, false, false);
+ buy(70, false, false);
+ buy(20, false, true); // fail
+ buy(5, false, false);
+
+ await().until(() -> orderManagerService.getOrders().size(), equalTo(3));
+ await().until(() -> creditService.getCredit(), equalTo(5));
+ }
+
+ @Test
+ public void testTotalCompensation() throws Exception {
+ // total credit is 100
+ for (int i = 0; i < 10; i++) {
+ if (i % 2 == 0) {
+ buy(10, false, false);
+ } else {
+ buy(10, true, true);
+ }
+ }
+
+ await().until(() -> orderManagerService.getOrders().size(), equalTo(5));
+ await().until(() -> creditService.getCredit(), equalTo(50));
+ }
+
+ private void buy(int amount, boolean failAtTheEnd, boolean shouldFail) {
+ try {
+ context.createFluentProducerTemplate()
+ .to("direct:saga")
+ .withHeader("amount", amount)
+ .withHeader("fail", failAtTheEnd)
+ .request();
+
+ if (shouldFail) {
+ fail("Exception not thrown");
+ }
+ } catch (Exception ex) {
+ if (!shouldFail) {
+ fail("Unexpected exception");
+ }
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ orderManagerService = new OrderManagerService();
+
+ creditService = new CreditService(100);
+
+ from("direct:saga")
+ .saga().propagation(SagaPropagation.REQUIRES_NEW)
+ .log("Creating a new order")
+ .to("direct:newOrder")
+ .log("Taking the credit")
+ .to("direct:reserveCredit")
+ .log("Finalizing")
+ .to("direct:finalize")
+ .log("Done!");
+
+
+ // Order service
+
+ from("direct:newOrder")
+ .saga()
+ .propagation(SagaPropagation.MANDATORY)
+ .compensation("direct:cancelOrder")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(orderManagerService, "newOrder")
+ .log("Order ${body} created");
+
+ from("direct:cancelOrder")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(orderManagerService, "cancelOrder")
+ .log("Order ${body} cancelled");
+
+
+ // Credit service
+
+ from("direct:reserveCredit")
+ .saga()
+ .propagation(SagaPropagation.MANDATORY)
+ .compensation("direct:refundCredit")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(creditService, "reserveCredit")
+ .log("Credit ${header.amount} reserved in action ${body}");
+
+ from("direct:refundCredit")
+ .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
+ .bean(creditService, "refundCredit")
+ .log("Credit for action ${body} refunded");
+
+
+ // Final actions
+ from("direct:finalize")
+ .saga().propagation(SagaPropagation.NOT_SUPPORTED)
+ .choice()
+ .when(header("fail").isEqualTo(true))
+ .process(x -> {
+ throw new RuntimeException("fail");
+ })
+ .end();
+
+ }
+ };
+ }
+
+ public static class OrderManagerService {
+
+ private Set<String> orders = new HashSet<>();
+
+ public synchronized void newOrder(String id) {
+ orders.add(id);
+ }
+
+ public synchronized void cancelOrder(String id) {
+ orders.remove(id);
+ }
+
+ public synchronized Set<String> getOrders() {
+ return new TreeSet<>(orders);
+ }
+ }
+
+ public static class CreditService {
+
+ private int totalCredit;
+
+ private Map<String, Integer> reservations = new HashMap<>();
+
+ public CreditService(int totalCredit) {
+ this.totalCredit = totalCredit;
+ }
+
+ public synchronized void reserveCredit(String id, @Header("amount") int amount) {
+ int credit = getCredit();
+ if (amount > credit) {
+ throw new IllegalStateException("Insufficient credit");
+ }
+ reservations.put(id, amount);
+ }
+
+ public synchronized void refundCredit(String id) {
+ reservations.remove(id);
+ }
+
+ public synchronized int getCredit() {
+ return totalCredit - reservations.values().stream().reduce(0, (a, b) -> a + b);
+ }
+ }
+}
diff --git a/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAFailuresIT.java b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAFailuresIT.java
new file mode 100644
index 0000000..2e09b8f
--- /dev/null
+++ b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAFailuresIT.java
@@ -0,0 +1,104 @@
+/**
+ * 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.service.lra;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.Ignore;
+import org.junit.Test;
+
+public class LRAFailuresIT extends AbstractLRATestSupport {
+
+ private AtomicInteger maxFailures;
+
+ @Test
+ @Ignore("https://issues.jboss.org/browse/JBTM-2978")
+ public void testCompensationAfterFailures() throws Exception {
+ maxFailures = new AtomicInteger(1);
+
+ MockEndpoint compensate = getMockEndpoint("mock:compensate");
+ compensate.expectedMessageCount(1);
+ compensate.setResultWaitTime(300000);
+
+ sendBody("direct:saga-compensate", "hello");
+
+ compensate.assertIsSatisfied();
+ }
+
+ @Test
+ @Ignore("https://issues.jboss.org/browse/JBTM-2977")
+ public void testCompletionAfterFailures() throws Exception {
+ maxFailures = new AtomicInteger(1);
+
+ MockEndpoint complete = getMockEndpoint("mock:complete");
+ complete.expectedMessageCount(1);
+ complete.setResultWaitTime(300000);
+
+ MockEndpoint end = getMockEndpoint("mock:end");
+ end.expectedBodiesReceived("hello");
+
+ sendBody("direct:saga-complete", "hello");
+
+ complete.assertIsSatisfied();
+ end.assertIsSatisfied();
+ }
+
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ from("direct:saga-compensate")
+ .saga()
+ .compensation("direct:compensate")
+ .process(x -> {
+ throw new RuntimeException("fail");
+ });
+
+ from("direct:saga-complete")
+ .saga()
+ .completion("direct:complete")
+ .to("mock:end");
+
+ from("direct:compensate")
+ .process(x -> {
+ int current = maxFailures.decrementAndGet();
+ if (current >= 0) {
+ throw new RuntimeException("compensation failure");
+ }
+ })
+ .to("mock:compensate");
+
+ from("direct:complete")
+ .process(x -> {
+ int current = maxFailures.decrementAndGet();
+ if (current >= 0) {
+ throw new RuntimeException("completion failure");
+ }
+ })
+ .to("mock:complete");
+
+ }
+ };
+ }
+
+}
diff --git a/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAManualIT.java b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAManualIT.java
new file mode 100644
index 0000000..b0213cd
--- /dev/null
+++ b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAManualIT.java
@@ -0,0 +1,95 @@
+/**
+ * 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.service.lra;
+
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.model.SagaCompletionMode;
+import org.junit.Ignore;
+import org.junit.Test;
+
+public class LRAManualIT extends AbstractLRATestSupport {
+
+ @Test
+ public void testCompletion() throws InterruptedException {
+ MockEndpoint completeEndpoint = getMockEndpoint("mock:complete");
+ completeEndpoint.expectedMessageCount(1);
+ completeEndpoint.expectedHeaderReceived("id", "1");
+
+ sendBody("direct:saga", "hello", Collections.singletonMap("myid", "1"));
+
+ completeEndpoint.assertIsSatisfied();
+ }
+
+ @Test
+ public void testFailure() throws InterruptedException {
+ MockEndpoint compensateEndpoint = getMockEndpoint("mock:compensate");
+ compensateEndpoint.expectedMessageCount(1);
+
+ sendBody("direct:saga", "fail");
+
+ compensateEndpoint.assertIsSatisfied();
+ }
+
+ @Test
+ @Ignore("https://issues.jboss.org/browse/JBTM-2979")
+ public void testTimeout() throws InterruptedException {
+ MockEndpoint compensateEndpoint = getMockEndpoint("mock:compensate");
+ compensateEndpoint.expectedMessageCount(1);
+
+ sendBody("direct:saga", "timeout");
+
+ compensateEndpoint.assertIsSatisfied();
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ from("direct:saga")
+ .saga()
+ .completionMode(SagaCompletionMode.MANUAL)
+ .timeout(1, TimeUnit.SECONDS)
+ .option("id", header("myid"))
+ .completion("direct:complete")
+ .compensation("direct:compensate")
+ .to("mock:endpoint")
+ .choice()
+ .when(body().isEqualTo("fail"))
+ .to("saga:compensate")
+ .when(body().isNotEqualTo("timeout"))
+ .to("saga:complete")
+ .end();
+
+ from("direct:complete")
+ .log("YES!")
+ .to("mock:complete");
+
+ from("direct:compensate")
+ .log("NO :(")
+ .to("mock:compensate");
+
+ }
+ };
+ }
+}
diff --git a/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAOptionsIT.java b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAOptionsIT.java
new file mode 100644
index 0000000..72e0a62
--- /dev/null
+++ b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAOptionsIT.java
@@ -0,0 +1,87 @@
+/**
+ * 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.service.lra;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class LRAOptionsIT extends AbstractLRATestSupport {
+
+ @Test
+ public void testHeaderForwardedToComplete() throws Exception {
+
+ MockEndpoint complete = getMockEndpoint("mock:complete");
+ complete.expectedMessageCount(1);
+ complete.expectedHeaderReceived("id", "myheader");
+ complete.expectedHeaderReceived("name", "Nicola");
+ complete.expectedMessagesMatches(ex -> ex.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION) != null);
+
+ template.sendBodyAndHeader("direct:workflow", "Hello", "myname", "Nicola");
+
+ complete.assertIsSatisfied();
+ }
+
+ @Test
+ public void testHeaderForwardedToCompensate() throws Exception {
+
+ MockEndpoint compensate = getMockEndpoint("mock:compensate");
+ compensate.expectedMessageCount(1);
+ compensate.expectedHeaderReceived("id", "myheader");
+ compensate.expectedHeaderReceived("name", "Nicola");
+ compensate.expectedMessagesMatches(ex -> ex.getIn().getHeader(Exchange.SAGA_LONG_RUNNING_ACTION) != null);
+
+ try {
+ template.sendBodyAndHeader("direct:workflow", "compensate", "myname", "Nicola");
+ Assert.fail("Should throw an exception");
+ } catch (Exception ex) {
+ // OK
+ }
+
+ compensate.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ from("direct:workflow")
+ .saga()
+ .option("id", constant("myheader"))
+ .option("name", header("myname"))
+ .completion("mock:complete")
+ .compensation("mock:compensate")
+ .choice()
+ .when(body().isEqualTo("compensate"))
+ .process(ex -> {
+ throw new RuntimeException("forced compensate");
+ })
+ .end()
+ .setHeader("myname", constant("TryToOverride"))
+ .setHeader("name", constant("TryToOverride"))
+ .to("mock:endpoint");
+
+ }
+ };
+ }
+
+}
diff --git a/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRATimeoutIT.java b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRATimeoutIT.java
new file mode 100644
index 0000000..858d03c
--- /dev/null
+++ b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRATimeoutIT.java
@@ -0,0 +1,94 @@
+/**
+ * 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.service.lra;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.model.SagaCompletionMode;
+import org.junit.Ignore;
+import org.junit.Test;
+
+public class LRATimeoutIT extends AbstractLRATestSupport {
+
+ @Test
+ @Ignore("https://issues.jboss.org/browse/JBTM-2979")
+ public void testTimeoutCalledCorrectly() throws Exception {
+
+ MockEndpoint compensate = getMockEndpoint("mock:compensate");
+ compensate.expectedMessageCount(1);
+ compensate.expectedHeaderReceived("id", "myid");
+
+ MockEndpoint end = getMockEndpoint("mock:end");
+ end.expectedMessageCount(1);
+
+ template.sendBody("direct:saga", "Hello");
+
+ end.assertIsSatisfied();
+ compensate.assertIsSatisfied();
+ }
+
+ @Test
+ public void testTimeoutHasNoEffectIfCompleted() throws Exception {
+
+ MockEndpoint compensate = getMockEndpoint("mock:compensate");
+ compensate.expectedMessageCount(1);
+ compensate.setResultWaitTime(500);
+
+ MockEndpoint complete = getMockEndpoint("mock:complete");
+ complete.expectedMessageCount(1);
+ complete.expectedHeaderReceived("id", "myid");
+
+ MockEndpoint end = getMockEndpoint("mock:end");
+ end.expectedMessageCount(1);
+
+ template.sendBody("direct:saga-auto", "Hello");
+
+ end.assertIsSatisfied();
+ complete.assertIsSatisfied();
+ compensate.assertIsNotSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ from("direct:saga")
+ .saga()
+ .timeout(100, TimeUnit.MILLISECONDS)
+ .option("id", constant("myid"))
+ .completionMode(SagaCompletionMode.MANUAL)
+ .compensation("mock:compensate")
+ .to("mock:end");
+
+ from("direct:saga-auto")
+ .saga()
+ .timeout(350, TimeUnit.MILLISECONDS)
+ .option("id", constant("myid"))
+ .compensation("mock:compensate")
+ .completion("mock:complete")
+ .to("mock:end");
+
+ }
+ };
+ }
+
+}
diff --git a/components/camel-lra/src/test/resources/log4j2.properties b/components/camel-lra/src/test/resources/log4j2.properties
new file mode 100644
index 0000000..a46962f
--- /dev/null
+++ b/components/camel-lra/src/test/resources/log4j2.properties
@@ -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.
+## ---------------------------------------------------------------------------
+
+appender.file.type = File
+appender.file.name = file
+appender.file.fileName = target/camel-lra-test.log
+appender.file.layout.type = PatternLayout
+appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+appender.out.type = Console
+appender.out.name = out
+appender.out.layout.type = PatternLayout
+appender.out.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+rootLogger.level = INFO
+rootLogger.appenderRef.file.ref = file
+#rootLogger.appenderRef.out.ref = out
\ No newline at end of file
diff --git a/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringSagaTest.java b/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringSagaTest.java
new file mode 100644
index 0000000..3d3c59e
--- /dev/null
+++ b/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringSagaTest.java
@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.spring.processor;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.component.mock.MockEndpoint;
+
+import static org.apache.camel.spring.processor.SpringTestHelper.createSpringCamelContext;
+
+public class SpringSagaTest extends ContextTestSupport {
+
+ public void testSendAMessageWithinASaga() throws Exception {
+ MockEndpoint resultEndpoint = getMockEndpoint("mock:end");
+ resultEndpoint.expectedBodiesReceived("correct");
+
+ MockEndpoint completionEndpoint = getMockEndpoint("mock:completion");
+ completionEndpoint.expectedMessageCount(1);
+ completionEndpoint.expectedHeaderReceived("myOptionKey", "myOptionValue");
+ completionEndpoint.expectedHeaderReceived("myOptionKey2", "myOptionValue2");
+
+ sendBody("direct:start", "correct");
+
+ resultEndpoint.assertIsSatisfied();
+ completionEndpoint.assertIsSatisfied();
+ }
+
+ public void testCompensationWithinASaga() throws Exception {
+ MockEndpoint resultEndpoint = getMockEndpoint("mock:end");
+ resultEndpoint.expectedMessageCount(1);
+ resultEndpoint.setResultWaitTime(100);
+
+ MockEndpoint compensationEndpoint = getMockEndpoint("mock:compensation");
+ compensationEndpoint.expectedMessageCount(1);
+ compensationEndpoint.expectedHeaderReceived("myOptionKey", "myOptionValue");
+ compensationEndpoint.expectedHeaderReceived("myOptionKey2", "myOptionValue2");
+
+ sendBody("direct:start", "fail");
+
+ compensationEndpoint .assertIsSatisfied();
+ resultEndpoint.assertIsNotSatisfied();
+ }
+
+ protected CamelContext createCamelContext() throws Exception {
+ return createSpringCamelContext(this, "org/apache/camel/spring/processor/saga.xml");
+ }
+}
diff --git a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/saga.xml b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/saga.xml
new file mode 100644
index 0000000..fdb2e9f
--- /dev/null
+++ b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/saga.xml
@@ -0,0 +1,57 @@
+<?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
+ ">
+
+ <bean id="sagaService" class="org.apache.camel.impl.saga.InMemorySagaService" init-method="start" destroy-method="stop" />
+
+ <!-- START SNIPPET: example -->
+ <camelContext xmlns="http://camel.apache.org/schema/spring">
+ <route>
+ <from uri="direct:start"/>
+ <saga>
+ <compensation uri="mock:compensation" />
+ <completion uri="mock:completion" />
+ <option optionName="myOptionKey">
+ <constant>myOptionValue</constant>
+ </option>
+ <option optionName="myOptionKey2">
+ <constant>myOptionValue2</constant>
+ </option>
+ </saga>
+ <choice>
+ <when>
+ <simple>${body} == 'fail'</simple>
+ <throwException exceptionType="java.lang.RuntimeException" message="fail" />
+ </when>
+ </choice>
+ <to uri="mock:end"/>
+ </route>
+
+
+
+ </camelContext>
+ <!-- END SNIPPET: example -->
+
+</beans>
diff --git a/components/pom.xml b/components/pom.xml
index d18aaa5..a789e55 100644
--- a/components/pom.xml
+++ b/components/pom.xml
@@ -198,6 +198,7 @@
<module>camel-ldif</module>
<module>camel-leveldb</module>
<module>camel-linkedin</module>
+ <module>camel-lra</module>
<module>camel-lucene</module>
<module>camel-lumberjack</module>
<module>camel-lzf</module>
diff --git a/components/readme.adoc b/components/readme.adoc
index 84d0c56..b314f3c 100644
--- a/components/readme.adoc
+++ b/components/readme.adoc
@@ -2,7 +2,7 @@ Components
^^^^^^^^^^
// components: START
-Number of Components: 286 in 197 JAR artifacts (19 deprecated)
+Number of Components: 287 in 197 JAR artifacts (19 deprecated)
[width="100%",cols="4,1,5",options="header"]
|===
@@ -686,6 +686,9 @@ Number of Components: 286 in 197 JAR artifacts (19 deprecated)
| link:camel-rss/src/main/docs/rss-component.adoc[RSS] (camel-rss) +
`rss:feedUri` | 2.0 | The rss component is used for consuming RSS feeds.
+| link:../camel-core/src/main/docs/saga-component.adoc[Saga] (camel-core) +
+`saga:action` | 2.21 | The default saga endpoint.
+
| link:camel-salesforce/camel-salesforce-component/src/main/docs/salesforce-component.adoc[Salesforce] (camel-salesforce) +
`salesforce:operationName:topicName` | 2.12 | The salesforce component is used for integrating Camel with the massive Salesforce API.
@@ -1044,7 +1047,7 @@ Miscellaneous Components
^^^^^^^^^^^^^^^^^^^^^^^^
// others: START
-Number of Miscellaneous Components: 38 in 38 JAR artifacts (13 deprecated)
+Number of Miscellaneous Components: 39 in 39 JAR artifacts (13 deprecated)
[width="100%",cols="4,1,5",options="header"]
|===
@@ -1076,6 +1079,8 @@ Number of Miscellaneous Components: 38 in 38 JAR artifacts (13 deprecated)
| link:camel-leveldb/src/main/docs/leveldb.adoc[LevelDB] (camel-leveldb) | 2.10 | Using LevelDB as persistent EIP store
+| link:camel-lra/src/main/docs/lra.adoc[Lra] (camel-lra) | 2.21 | Camel saga binding for Long-Running-Action framework
+
| link:camel-opentracing/src/main/docs/opentracing.adoc[OpenTracing] (camel-opentracing) | 2.19 | Distributed tracing using OpenTracing
| link:camel-reactor/src/main/docs/reactor.adoc[Reactor] (camel-reactor) | 2.20 | Reactor based back-end for Camel's reactive streams component
diff --git a/docs/user-manual/en/SUMMARY.md b/docs/user-manual/en/SUMMARY.md
index 9e38e87..ddec6de 100644
--- a/docs/user-manual/en/SUMMARY.md
+++ b/docs/user-manual/en/SUMMARY.md
@@ -104,6 +104,7 @@
* [Ref](ref-component.adoc)
* [REST](rest-component.adoc)
* [REST API](rest-api-component.adoc)
+ * [Saga](saga-component.adoc)
* [Scheduler](scheduler-component.adoc)
* [SEDA](seda-component.adoc)
* [Stub](stub-component.adoc)
@@ -405,6 +406,7 @@
* [Jasypt](jasypt.adoc)
* [Kura](kura.adoc)
* [LevelDB](leveldb.adoc)
+ * [Lra](lra.adoc)
* [OpenTracing](opentracing.adoc)
* [Reactor](reactor.adoc)
* [Ribbon](ribbon.adoc)
diff --git a/parent/pom.xml b/parent/pom.xml
index c0374e3..886e4eb 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -1577,6 +1577,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-lra</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-lucene</artifactId>
<version>${project.version}</version>
</dependency>
@@ -3011,6 +3016,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-lra-starter</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-lucene-starter</artifactId>
<version>${project.version}</version>
</dependency>
diff --git a/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentAutoConfiguration.java b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentAutoConfiguration.java
new file mode 100644
index 0000000..c9c90ed
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentAutoConfiguration.java
@@ -0,0 +1,128 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.saga.springboot;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Generated;
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.saga.SagaComponent;
+import org.apache.camel.spi.ComponentCustomizer;
+import org.apache.camel.spi.HasId;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.spring.boot.ComponentConfigurationProperties;
+import org.apache.camel.spring.boot.util.CamelPropertiesHelper;
+import org.apache.camel.spring.boot.util.ConditionalOnCamelContextAndAutoConfigurationBeans;
+import org.apache.camel.spring.boot.util.GroupCondition;
+import org.apache.camel.spring.boot.util.HierarchicalPropertiesEvaluator;
+import org.apache.camel.util.IntrospectionSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+
+/**
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.SpringBootAutoConfigurationMojo")
+@Configuration
+@Conditional({ConditionalOnCamelContextAndAutoConfigurationBeans.class,
+ SagaComponentAutoConfiguration.GroupConditions.class})
+@AutoConfigureAfter(CamelAutoConfiguration.class)
+@EnableConfigurationProperties({ComponentConfigurationProperties.class,
+ SagaComponentConfiguration.class})
+public class SagaComponentAutoConfiguration {
+
+ private static final Logger LOGGER = LoggerFactory
+ .getLogger(SagaComponentAutoConfiguration.class);
+ @Autowired
+ private ApplicationContext applicationContext;
+ @Autowired
+ private CamelContext camelContext;
+ @Autowired
+ private SagaComponentConfiguration configuration;
+ @Autowired(required = false)
+ private List<ComponentCustomizer<SagaComponent>> customizers;
+
+ static class GroupConditions extends GroupCondition {
+ public GroupConditions() {
+ super("camel.component", "camel.component.saga");
+ }
+ }
+
+ @Lazy
+ @Bean(name = "saga-component")
+ @ConditionalOnMissingBean(SagaComponent.class)
+ public SagaComponent configureSagaComponent() throws Exception {
+ SagaComponent component = new SagaComponent();
+ component.setCamelContext(camelContext);
+ Map<String, Object> parameters = new HashMap<>();
+ IntrospectionSupport.getProperties(configuration, parameters, null,
+ false);
+ for (Map.Entry<String, Object> entry : parameters.entrySet()) {
+ Object value = entry.getValue();
+ Class<?> paramClass = value.getClass();
+ if (paramClass.getName().endsWith("NestedConfiguration")) {
+ Class nestedClass = null;
+ try {
+ nestedClass = (Class) paramClass.getDeclaredField(
+ "CAMEL_NESTED_CLASS").get(null);
+ HashMap<String, Object> nestedParameters = new HashMap<>();
+ IntrospectionSupport.getProperties(value, nestedParameters,
+ null, false);
+ Object nestedProperty = nestedClass.newInstance();
+ CamelPropertiesHelper.setCamelProperties(camelContext,
+ nestedProperty, nestedParameters, false);
+ entry.setValue(nestedProperty);
+ } catch (NoSuchFieldException e) {
+ }
+ }
+ }
+ CamelPropertiesHelper.setCamelProperties(camelContext, component,
+ parameters, false);
+ if (ObjectHelper.isNotEmpty(customizers)) {
+ for (ComponentCustomizer<SagaComponent> customizer : customizers) {
+ boolean useCustomizer = (customizer instanceof HasId)
+ ? HierarchicalPropertiesEvaluator.evaluate(
+ applicationContext.getEnvironment(),
+ "camel.component.customizer",
+ "camel.component.saga.customizer",
+ ((HasId) customizer).getId())
+ : HierarchicalPropertiesEvaluator.evaluate(
+ applicationContext.getEnvironment(),
+ "camel.component.customizer",
+ "camel.component.saga.customizer");
+ if (useCustomizer) {
+ LOGGER.debug("Configure component {}, with customizer {}",
+ component, customizer);
+ customizer.customize(component);
+ }
+ }
+ }
+ return component;
+ }
+}
\ No newline at end of file
diff --git a/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentConfiguration.java
new file mode 100644
index 0000000..bc67138
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentConfiguration.java
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.saga.springboot;
+
+import javax.annotation.Generated;
+import org.apache.camel.spring.boot.ComponentConfigurationPropertiesCommon;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * The default saga endpoint.
+ *
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.SpringBootAutoConfigurationMojo")
+@ConfigurationProperties(prefix = "camel.component.saga")
+public class SagaComponentConfiguration
+ extends
+ ComponentConfigurationPropertiesCommon {
+
+ /**
+ * Whether the component should resolve property placeholders on itself when
+ * starting. Only properties which are of String type can use property
+ * placeholders.
+ */
+ private Boolean resolvePropertyPlaceholders = true;
+
+ public Boolean getResolvePropertyPlaceholders() {
+ return resolvePropertyPlaceholders;
+ }
+
+ public void setResolvePropertyPlaceholders(
+ Boolean resolvePropertyPlaceholders) {
+ this.resolvePropertyPlaceholders = resolvePropertyPlaceholders;
+ }
+}
\ No newline at end of file
diff --git a/platforms/spring-boot/components-starter/camel-core-starter/src/main/resources/META-INF/spring.factories b/platforms/spring-boot/components-starter/camel-core-starter/src/main/resources/META-INF/spring.factories
index 904d89c..ecdb8e2 100644
--- a/platforms/spring-boot/components-starter/camel-core-starter/src/main/resources/META-INF/spring.factories
+++ b/platforms/spring-boot/components-starter/camel-core-starter/src/main/resources/META-INF/spring.factories
@@ -55,5 +55,7 @@ org.apache.camel.component.binding.springboot.BindingNameComponentAutoConfigurat
org.apache.camel.component.bean.springboot.BeanComponentAutoConfiguration,\
org.apache.camel.component.dataformat.springboot.DataFormatComponentAutoConfiguration,\
org.apache.camel.component.rest.springboot.RestApiComponentAutoConfiguration,\
-org.apache.camel.model.rest.springboot.RestConfigurationDefinitionAutoConfiguration
+org.apache.camel.model.rest.springboot.RestConfigurationDefinitionAutoConfiguration,\
+org.apache.camel.component.saga.springboot.SagaComponentAutoConfiguration
+
diff --git a/platforms/spring-boot/components-starter/camel-lra-starter/pom.xml b/platforms/spring-boot/components-starter/camel-lra-starter/pom.xml
new file mode 100644
index 0000000..abef786
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-lra-starter/pom.xml
@@ -0,0 +1,53 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components-starter</artifactId>
+ <version>2.21.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>camel-lra-starter</artifactId>
+ <packaging>jar</packaging>
+ <name>Spring-Boot Starter :: Camel :: Long-Running-Action</name>
+ <description>Spring-Boot Starter for Camel saga binding for Long-Running-Action framework</description>
+ <dependencies>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter</artifactId>
+ <version>${spring-boot-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-lra</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <!--START OF GENERATED CODE-->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core-starter</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-spring-boot-starter</artifactId>
+ </dependency>
+ <!--END OF GENERATED CODE-->
+ </dependencies>
+</project>
diff --git a/platforms/spring-boot/components-starter/camel-lra-starter/src/main/java/org/apache/camel/service/lra/springboot/LraServiceAutoConfiguration.java b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/java/org/apache/camel/service/lra/springboot/LraServiceAutoConfiguration.java
new file mode 100644
index 0000000..8bdd273
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/java/org/apache/camel/service/lra/springboot/LraServiceAutoConfiguration.java
@@ -0,0 +1,77 @@
+/**
+ * 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.service.lra.springboot;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.saga.CamelSagaService;
+import org.apache.camel.service.lra.LRASagaService;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.spring.boot.util.CamelPropertiesHelper;
+import org.apache.camel.spring.boot.util.ConditionalOnCamelContextAndAutoConfigurationBeans;
+import org.apache.camel.spring.boot.util.GroupCondition;
+import org.apache.camel.util.IntrospectionSupport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@Conditional({ConditionalOnCamelContextAndAutoConfigurationBeans.class,
+ LraServiceAutoConfiguration.GroupConditions.class})
+@AutoConfigureAfter(CamelAutoConfiguration.class)
+@EnableConfigurationProperties({LraServiceConfiguration.class})
+public class LraServiceAutoConfiguration {
+
+ private static final Logger LOGGER = LoggerFactory
+ .getLogger(LraServiceAutoConfiguration.class);
+
+ @Autowired
+ private CamelContext camelContext;
+
+ @Autowired
+ private LraServiceConfiguration configuration;
+
+
+ static class GroupConditions extends GroupCondition {
+ public GroupConditions() {
+ super("camel.service", "camel.service.lra");
+ }
+ }
+
+ @Bean(name = "lra-service")
+ @ConditionalOnMissingBean(CamelSagaService.class)
+ @ConditionalOnProperty(value = "camel.service.lra.enabled", havingValue = "true")
+ public LRASagaService configureLraSagaService() throws Exception {
+ LRASagaService service = new LRASagaService();
+
+ Map<String, Object> parameters = new HashMap<>();
+ IntrospectionSupport.getProperties(configuration, parameters, null, false);
+ CamelPropertiesHelper.setCamelProperties(camelContext, service, parameters, false);
+
+ camelContext.addService(service);
+ return service;
+ }
+}
\ No newline at end of file
diff --git a/platforms/spring-boot/components-starter/camel-lra-starter/src/main/java/org/apache/camel/service/lra/springboot/LraServiceConfiguration.java b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/java/org/apache/camel/service/lra/springboot/LraServiceConfiguration.java
new file mode 100644
index 0000000..77eedc5
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/java/org/apache/camel/service/lra/springboot/LraServiceConfiguration.java
@@ -0,0 +1,94 @@
+/**
+ * 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.service.lra.springboot;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import static org.apache.camel.service.lra.LRAConstants.DEFAULT_COORDINATOR_CONTEXT_PATH;
+import static org.apache.camel.service.lra.LRAConstants.DEFAULT_LOCAL_PARTICIPANT_CONTEXT_PATH;
+
+/**
+ * Spring-boot Auto-configuration for LRA service.
+ */
+@ConfigurationProperties(prefix = "camel.service.lra")
+public class LraServiceConfiguration {
+
+ /**
+ * Global option to enable/disable component auto-configuration, default is true.
+ */
+ private boolean enabled = true;
+
+ /**
+ * The base URL of the LRA coordinator service (e.g. http://lra-host:8080)
+ */
+ private String coordinatorUrl;
+
+ /**
+ * The context path of the LRA coordinator service
+ */
+ private String coordinatorContextPath = DEFAULT_COORDINATOR_CONTEXT_PATH;
+
+ /**
+ * The local URL where the coordinator should send callbacks to (e.g. http://my-host-name:8080)
+ */
+ private String localParticipantUrl;
+
+ /**
+ * The context path of the local participant callback services
+ */
+ private String localParticipantContextPath = DEFAULT_LOCAL_PARTICIPANT_CONTEXT_PATH;
+
+ public boolean isEnabled() {
+ return enabled;
+ }
+
+ public void setEnabled(boolean enabled) {
+ this.enabled = enabled;
+ }
+
+ public String getCoordinatorUrl() {
+ return coordinatorUrl;
+ }
+
+ public void setCoordinatorUrl(String coordinatorUrl) {
+ this.coordinatorUrl = coordinatorUrl;
+ }
+
+ public String getCoordinatorContextPath() {
+ return coordinatorContextPath;
+ }
+
+ public void setCoordinatorContextPath(String coordinatorContextPath) {
+ this.coordinatorContextPath = coordinatorContextPath;
+ }
+
+ public String getLocalParticipantUrl() {
+ return localParticipantUrl;
+ }
+
+ public void setLocalParticipantUrl(String localParticipantUrl) {
+ this.localParticipantUrl = localParticipantUrl;
+ }
+
+ public String getLocalParticipantContextPath() {
+ return localParticipantContextPath;
+ }
+
+ public void setLocalParticipantContextPath(String localParticipantContextPath) {
+ this.localParticipantContextPath = localParticipantContextPath;
+ }
+}
\ No newline at end of file
diff --git a/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/LICENSE.txt b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/LICENSE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/LICENSE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed 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.
+
diff --git a/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/NOTICE.txt b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/NOTICE.txt
new file mode 100644
index 0000000..2e215bf
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/NOTICE.txt
@@ -0,0 +1,11 @@
+ =========================================================================
+ == NOTICE file corresponding to the section 4 d of ==
+ == the Apache License, Version 2.0, ==
+ == in this case for the Apache Camel distribution. ==
+ =========================================================================
+
+ This product includes software developed by
+ The Apache Software Foundation (http://www.apache.org/).
+
+ Please read the different LICENSE files present in the licenses directory of
+ this distribution.
diff --git a/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/spring.factories b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/spring.factories
new file mode 100644
index 0000000..4332dc1
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,18 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.apache.camel.service.lra.springboot.LraServiceAutoConfiguration
diff --git a/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/spring.provides b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/spring.provides
new file mode 100644
index 0000000..b4b86fe
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-lra-starter/src/main/resources/META-INF/spring.provides
@@ -0,0 +1,17 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+provides: camel-lra
diff --git a/platforms/spring-boot/components-starter/pom.xml b/platforms/spring-boot/components-starter/pom.xml
index 8ec568f..a6a4d22 100644
--- a/platforms/spring-boot/components-starter/pom.xml
+++ b/platforms/spring-boot/components-starter/pom.xml
@@ -215,6 +215,7 @@
<module>camel-ldif-starter</module>
<module>camel-leveldb-starter</module>
<module>camel-linkedin-starter</module>
+ <module>camel-lra-starter</module>
<module>camel-lucene-starter</module>
<module>camel-lumberjack-starter</module>
<module>camel-lzf-starter</module>
diff --git a/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml b/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml
index f31837a..68a0c19 100644
--- a/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml
+++ b/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml
@@ -1751,6 +1751,16 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-lra</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-lra-starter</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-lucene</artifactId>
<version>${project.version}</version>
</dependency>
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.
[camel] 02/02: CAMEL-11665: fix configuration and documentation
Posted by nf...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nferraro pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit fd1fbd3bf7f1d6bca9b8c30703645dcb3f4c1f55
Author: Nicola Ferraro <ni...@gmail.com>
AuthorDate: Thu Jan 11 08:18:32 2018 +0100
CAMEL-11665: fix configuration and documentation
---
camel-core/readme.adoc | 2 +-
camel-core/src/main/docs/saga-component.adoc | 2 --
camel-core/src/main/java/org/apache/camel/Exchange.java | 3 ++-
.../java/org/apache/camel/component/saga/SagaEndpoint.java | 4 ++--
.../org/apache/camel/model/SagaActionUriDefinition.java | 2 +-
components/readme.adoc | 13 ++++++++++---
.../saga/springboot/SagaComponentConfiguration.java | 3 ++-
7 files changed, 18 insertions(+), 11 deletions(-)
diff --git a/camel-core/readme.adoc b/camel-core/readme.adoc
index 50eee0e..d84141c 100644
--- a/camel-core/readme.adoc
+++ b/camel-core/readme.adoc
@@ -64,7 +64,7 @@ Number of Components: 26 in 1 JAR artifacts (1 deprecated)
`rest-api:path/contextIdPattern` | 2.16 | The rest-api component is used for providing Swagger API of the REST services which has been defined using the rest-dsl in Camel.
| link:src/main/docs/saga-component.adoc[Saga] (camel-core) +
-`saga:action` | 2.21 | The default saga endpoint.
+`saga:action` | 2.21 | The saga component provides access to advanced options for managing the flow in the Saga EIP.
| link:src/main/docs/scheduler-component.adoc[Scheduler] (camel-core) +
`scheduler:name` | 2.15 | The scheduler component is used for generating message exchanges when a scheduler fires.
diff --git a/camel-core/src/main/docs/saga-component.adoc b/camel-core/src/main/docs/saga-component.adoc
index 2610f4b..b0ead4c 100644
--- a/camel-core/src/main/docs/saga-component.adoc
+++ b/camel-core/src/main/docs/saga-component.adoc
@@ -16,8 +16,6 @@ Refer to the Saga EIP documentation for help on using sagas in common scenarios.
saga:action
----
-Where *`action`* can be ...
-
=== Options
// component options: START
diff --git a/camel-core/src/main/java/org/apache/camel/Exchange.java b/camel-core/src/main/java/org/apache/camel/Exchange.java
index f3e3774..3ec0fea 100644
--- a/camel-core/src/main/java/org/apache/camel/Exchange.java
+++ b/camel-core/src/main/java/org/apache/camel/Exchange.java
@@ -170,7 +170,8 @@ public interface Exchange {
String LOOP_INDEX = "CamelLoopIndex";
String LOOP_SIZE = "CamelLoopSize";
- // Long running action (saga)
+ // Long running action (saga): using "Long-Running-Action" as header value allows sagas
+ // to be propagated to any remote system supporting the LRA framework
String SAGA_LONG_RUNNING_ACTION = "Long-Running-Action";
String MAXIMUM_CACHE_POOL_SIZE = "CamelMaximumCachePoolSize";
diff --git a/camel-core/src/main/java/org/apache/camel/component/saga/SagaEndpoint.java b/camel-core/src/main/java/org/apache/camel/component/saga/SagaEndpoint.java
index 96cd167..a0f3114 100644
--- a/camel-core/src/main/java/org/apache/camel/component/saga/SagaEndpoint.java
+++ b/camel-core/src/main/java/org/apache/camel/component/saga/SagaEndpoint.java
@@ -26,7 +26,7 @@ import org.apache.camel.spi.UriPath;
import org.apache.camel.util.ObjectHelper;
/**
- * The default saga endpoint.
+ * The saga component provides access to advanced options for managing the flow in the Saga EIP.
*/
@UriEndpoint(firstVersion = "2.21.0", scheme = "saga", title = "Saga", syntax = "saga:action", producerOnly = true, label = "core,endpoint")
public class SagaEndpoint extends DefaultEndpoint {
@@ -63,7 +63,7 @@ public class SagaEndpoint extends DefaultEndpoint {
@Override
public boolean isSingleton() {
- return false;
+ return true;
}
}
diff --git a/camel-core/src/main/java/org/apache/camel/model/SagaActionUriDefinition.java b/camel-core/src/main/java/org/apache/camel/model/SagaActionUriDefinition.java
index 39a080e..cb3cefa 100644
--- a/camel-core/src/main/java/org/apache/camel/model/SagaActionUriDefinition.java
+++ b/camel-core/src/main/java/org/apache/camel/model/SagaActionUriDefinition.java
@@ -24,7 +24,7 @@ import org.apache.camel.spi.Metadata;
import org.apache.camel.util.ObjectHelper;
/**
- * Allows to declare saga actions e.g. to complete or compensate a saga
+ * Allows to declare saga actions to complete or compensate a saga
*
*/
@Metadata(label = "eip,routing")
diff --git a/components/readme.adoc b/components/readme.adoc
index b314f3c..8847e47 100644
--- a/components/readme.adoc
+++ b/components/readme.adoc
@@ -2,7 +2,7 @@ Components
^^^^^^^^^^
// components: START
-Number of Components: 287 in 197 JAR artifacts (19 deprecated)
+Number of Components: 288 in 197 JAR artifacts (19 deprecated)
[width="100%",cols="4,1,5",options="header"]
|===
@@ -80,6 +80,9 @@ Number of Components: 287 in 197 JAR artifacts (19 deprecated)
| link:camel-aws/src/main/docs/aws-lambda-component.adoc[AWS Lambda] (camel-aws) +
`aws-lambda:function` | 2.20 | The aws-lambda is used for managing and invoking functions from Amazon Lambda.
+| link:camel-aws/src/main/docs/aws-mq-component.adoc[AWS MQ] (camel-aws) +
+`aws-mq:label` | 2.21 | The aws-mq is used for managing Amazon MQ instances.
+
| link:camel-aws/src/main/docs/aws-s3-component.adoc[AWS S3 Storage Service] (camel-aws) +
`aws-s3:bucketNameOrArn` | 2.8 | The aws-s3 component is used for storing and retrieving objecct from Amazon S3 Storage Service.
@@ -687,7 +690,7 @@ Number of Components: 287 in 197 JAR artifacts (19 deprecated)
`rss:feedUri` | 2.0 | The rss component is used for consuming RSS feeds.
| link:../camel-core/src/main/docs/saga-component.adoc[Saga] (camel-core) +
-`saga:action` | 2.21 | The default saga endpoint.
+`saga:action` | 2.21 | The saga component provides access to advanced options for managing the flow in the Saga EIP.
| link:camel-salesforce/camel-salesforce-component/src/main/docs/salesforce-component.adoc[Salesforce] (camel-salesforce) +
`salesforce:operationName:topicName` | 2.12 | The salesforce component is used for integrating Camel with the massive Salesforce API.
@@ -1047,7 +1050,7 @@ Miscellaneous Components
^^^^^^^^^^^^^^^^^^^^^^^^
// others: START
-Number of Miscellaneous Components: 39 in 39 JAR artifacts (13 deprecated)
+Number of Miscellaneous Components: 41 in 41 JAR artifacts (14 deprecated)
[width="100%",cols="4,1,5",options="header"]
|===
@@ -1101,10 +1104,14 @@ Number of Miscellaneous Components: 39 in 39 JAR artifacts (13 deprecated)
| link:camel-spring-boot/src/main/docs/spring-boot.adoc[Spring Boot] (camel-spring-boot) | 2.15 | Using Camel with Spring Boot
+| link:camel-spring-boot2/src/main/docs/spring-boot2.adoc[Spring Boot2] (camel-spring-boot2) | 2.20 | Using Camel with Spring Boot 2 (work in progress)
+
| link:camel-spring-cloud/src/main/docs/spring-cloud.adoc[Spring Cloud] (camel-spring-cloud) | 2.19 | Camel Cloud integration with Spring Cloud
| link:camel-spring-cloud-netflix/src/main/docs/spring-cloud-netflix.adoc[Spring Cloud Netflix] (camel-spring-cloud-netflix) | 2.19 | Camel Cloud integration with Spring Cloud Netflix
+| link:camel-spring-dm/src/main/docs/spring-dm.adoc[Spring DM] (camel-spring-dm) | 2.18 | *deprecated* Camel SpringDM (OSGi) XML DSL
+
| link:camel-spring-javaconfig/src/main/docs/spring-javaconfig.adoc[Spring Java Configuration] (camel-spring-javaconfig) | 2.0 | Using Camel with Spring Java Configuration
| link:camel-spring-security/src/main/docs/spring-security.adoc[Spring Security] (camel-spring-security) | 2.3 | Security using Spring Security
diff --git a/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentConfiguration.java
index bc67138..7050359 100644
--- a/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-core-starter/src/main/java/org/apache/camel/component/saga/springboot/SagaComponentConfiguration.java
@@ -21,7 +21,8 @@ import org.apache.camel.spring.boot.ComponentConfigurationPropertiesCommon;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
- * The default saga endpoint.
+ * The saga component provides access to advanced options for managing the flow
+ * in the Saga EIP.
*
* Generated by camel-package-maven-plugin - do not edit this file!
*/
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.