You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2020/08/26 09:41:47 UTC
[camel] branch master updated (80e842b -> c9c1c1e)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from 80e842b CAMEL-15413: RouteBuilderConfigurer - Rename to LambdaRouteBuilder
new 0d75483 Added camel-redis component and RedisAggregationRepository using Redisson
new c9c1c1e CAMEL-15468: camel-redis - Polished and rebuild
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 | 1 +
bom/camel-bom/pom.xml | 5 +
camel-dependencies/pom.xml | 1 +
.../org/apache/camel/catalog/docs.properties | 1 +
.../org/apache/camel/catalog/docs/redis.adoc | 11 +
.../org/apache/camel/catalog/others.properties | 1 +
.../org/apache/camel/catalog/others/redis.json | 15 +
components/camel-redis/pom.xml | 68 ++++
.../services/org/apache/camel/other.properties | 7 +
.../camel-redis/src/generated/resources/redis.json | 15 +
components/camel-redis/src/main/docs/redis.adoc | 11 +
.../aggregate/RedisAggregationRepository.java | 382 +++++++++++++++++++++
.../processor/aggregate/AggregateRedisTest.java | 67 ++++
.../processor/aggregate/MyAggregationStrategy.java | 45 +++
.../src/test/resources/log4j2.properties | 28 ++
components/pom.xml | 1 +
docs/components/modules/others/nav.adoc | 1 +
docs/components/modules/others/pages/redis.adoc | 13 +
parent/pom.xml | 6 +
19 files changed, 679 insertions(+)
create mode 100644 catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/redis.adoc
create mode 100644 catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others/redis.json
create mode 100644 components/camel-redis/pom.xml
create mode 100644 components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/other.properties
create mode 100644 components/camel-redis/src/generated/resources/redis.json
create mode 100644 components/camel-redis/src/main/docs/redis.adoc
create mode 100644 components/camel-redis/src/main/java/org/apache/camel/component/redis/processor/aggregate/RedisAggregationRepository.java
create mode 100644 components/camel-redis/src/test/java/org/apache/camel/component/redis/processor/aggregate/AggregateRedisTest.java
create mode 100644 components/camel-redis/src/test/java/org/apache/camel/component/redis/processor/aggregate/MyAggregationStrategy.java
create mode 100644 components/camel-redis/src/test/resources/log4j2.properties
create mode 100644 docs/components/modules/others/pages/redis.adoc
[camel] 02/02: CAMEL-15468: camel-redis - Polished and rebuild
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit c9c1c1e85c58a021eb76002d08e6cc16ace879a4
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Aug 26 11:08:19 2020 +0200
CAMEL-15468: camel-redis - Polished and rebuild
---
apache-camel/src/main/descriptors/common-bin.xml | 1 +
.../org/apache/camel/catalog/docs.properties | 1 +
.../org/apache/camel/catalog/docs/redis.adoc | 11 +
.../org/apache/camel/catalog/others.properties | 1 +
.../org/apache/camel/catalog/others/redis.json | 15 ++
components/camel-redis/pom.xml | 42 ++--
.../services/org/apache/camel/other.properties | 7 +
.../camel-redis/src/generated/resources/redis.json | 15 ++
components/camel-redis/src/main/docs/redis.adoc | 11 +
.../aggregate/RedisAggregationRepository.java | 254 ++++++++++++---------
.../camel-redis/src/test/AggregateRedisTest.java | 51 -----
.../processor/aggregate/AggregateRedisTest.java | 67 ++++++
.../aggregate}/MyAggregationStrategy.java | 21 +-
.../src/test/resources/log4j2.properties | 28 +++
docs/components/modules/others/nav.adoc | 1 +
docs/components/modules/others/pages/redis.adoc | 13 ++
parent/pom.xml | 2 +-
17 files changed, 353 insertions(+), 188 deletions(-)
diff --git a/apache-camel/src/main/descriptors/common-bin.xml b/apache-camel/src/main/descriptors/common-bin.xml
index 827a5f9..a69243c 100644
--- a/apache-camel/src/main/descriptors/common-bin.xml
+++ b/apache-camel/src/main/descriptors/common-bin.xml
@@ -307,6 +307,7 @@
<include>org.apache.camel:camel-reactive-executor-vertx</include>
<include>org.apache.camel:camel-reactive-streams</include>
<include>org.apache.camel:camel-reactor</include>
+ <include>org.apache.camel:camel-redis</include>
<include>org.apache.camel:camel-ref</include>
<include>org.apache.camel:camel-resilience4j</include>
<include>org.apache.camel:camel-rest</include>
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties
index 88166b3..b70630c 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties
@@ -395,6 +395,7 @@ reactive-executor-vertx
reactive-streams-component
reactor
recipientList-eip
+redis
ref-component
ref-language
removeHeader-eip
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/redis.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/redis.adoc
new file mode 100644
index 0000000..1e81d98
--- /dev/null
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/redis.adoc
@@ -0,0 +1,11 @@
+[[redis-component]]
+= Redis Component
+:docTitle: Redis
+:artifactId: camel-redis
+:description: Aggregation repository using Redis as datastore
+:since: 3.5
+:supportLevel: Preview
+
+*Since Camel {since}*
+
+The Redis component provides an `AggregationStrategy` to use Redis as the backend datastore.
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties
index f4a9fc2..93ceff4 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others.properties
@@ -17,6 +17,7 @@ opentracing
platform-http-vertx
reactive-executor-vertx
reactor
+redis
resilience4j
ribbon
rxjava
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others/redis.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others/redis.json
new file mode 100644
index 0000000..6bbfb91
--- /dev/null
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others/redis.json
@@ -0,0 +1,15 @@
+{
+ "other": {
+ "kind": "other",
+ "name": "redis",
+ "title": "Redis",
+ "description": "Aggregation repository using Redis as datastore",
+ "deprecated": false,
+ "firstVersion": "3.5.0",
+ "label": "database",
+ "supportLevel": "Preview",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-redis",
+ "version": "3.5.0-SNAPSHOT"
+ }
+}
diff --git a/components/camel-redis/pom.xml b/components/camel-redis/pom.xml
index 700269e..50d73cb 100644
--- a/components/camel-redis/pom.xml
+++ b/components/camel-redis/pom.xml
@@ -17,7 +17,8 @@
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">
+<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>
@@ -30,35 +31,38 @@
<artifactId>camel-redis</artifactId>
<packaging>jar</packaging>
<name>Camel :: Redis</name>
- <description>Camel Redis support</description>
+ <description>Aggregation repository using Redis as datastore</description>
+
+ <properties>
+ <firstVersion>3.5.0</firstVersion>
+ <label>database</label>
+ </properties>
<dependencies>
<dependency>
- <groupId>org.redisson</groupId>
- <artifactId>redisson</artifactId>
- <version>${redisson-version}</version>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-support</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.redisson</groupId>
+ <artifactId>redisson</artifactId>
+ <version>${redisson-version}</version>
</dependency>
+ <!-- testing -->
<dependency>
<groupId>org.apache.camel</groupId>
- <artifactId>camel-test-spring-junit5</artifactId>
+ <artifactId>camel-test-junit5</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
- <build>
- <plugins>
- <plugin>
- <artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <systemPropertyVariables>
- <endpoint>${endpoint}</endpoint>
- </systemPropertyVariables>
- </configuration>
- </plugin>
- </plugins>
- </build>
-
</project>
diff --git a/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/other.properties b/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/other.properties
new file mode 100644
index 0000000..fc3636e
--- /dev/null
+++ b/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/other.properties
@@ -0,0 +1,7 @@
+# Generated by camel build tools - do NOT edit this file!
+name=redis
+groupId=org.apache.camel
+artifactId=camel-redis
+version=3.5.0-SNAPSHOT
+projectName=Camel :: Redis
+projectDescription=Aggregation repository using Redis as datastore
diff --git a/components/camel-redis/src/generated/resources/redis.json b/components/camel-redis/src/generated/resources/redis.json
new file mode 100644
index 0000000..6bbfb91
--- /dev/null
+++ b/components/camel-redis/src/generated/resources/redis.json
@@ -0,0 +1,15 @@
+{
+ "other": {
+ "kind": "other",
+ "name": "redis",
+ "title": "Redis",
+ "description": "Aggregation repository using Redis as datastore",
+ "deprecated": false,
+ "firstVersion": "3.5.0",
+ "label": "database",
+ "supportLevel": "Preview",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-redis",
+ "version": "3.5.0-SNAPSHOT"
+ }
+}
diff --git a/components/camel-redis/src/main/docs/redis.adoc b/components/camel-redis/src/main/docs/redis.adoc
new file mode 100644
index 0000000..1e81d98
--- /dev/null
+++ b/components/camel-redis/src/main/docs/redis.adoc
@@ -0,0 +1,11 @@
+[[redis-component]]
+= Redis Component
+:docTitle: Redis
+:artifactId: camel-redis
+:description: Aggregation repository using Redis as datastore
+:since: 3.5
+:supportLevel: Preview
+
+*Since Camel {since}*
+
+The Redis component provides an `AggregationStrategy` to use Redis as the backend datastore.
diff --git a/components/camel-redis/src/main/java/org/apache/camel/component/redis/processor/aggregate/RedisAggregationRepository.java b/components/camel-redis/src/main/java/org/apache/camel/component/redis/processor/aggregate/RedisAggregationRepository.java
index 4b5ab34..49934b4 100644
--- a/components/camel-redis/src/main/java/org/apache/camel/component/redis/processor/aggregate/RedisAggregationRepository.java
+++ b/components/camel-redis/src/main/java/org/apache/camel/component/redis/processor/aggregate/RedisAggregationRepository.java
@@ -14,27 +14,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.camel.component.redis.processor.aggregate;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
-import java.util.TreeSet;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.spi.OptimisticLockingAggregationRepository;
import org.apache.camel.spi.RecoverableAggregationRepository;
-import org.apache.camel.spi.OptimisticLockingAggregationRepository.OptimisticLockingException;
import org.apache.camel.support.DefaultExchange;
import org.apache.camel.support.DefaultExchangeHolder;
import org.apache.camel.support.service.ServiceSupport;
@@ -49,52 +39,57 @@ import org.redisson.config.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * {@link org.apache.camel.spi.AggregationRepository} using Redis as store.
+ */
public class RedisAggregationRepository extends ServiceSupport
- implements RecoverableAggregationRepository, OptimisticLockingAggregationRepository {
- private static final Logger LOG = LoggerFactory.getLogger(RedisAggregationRepository.class);
- private static final String COMPLETED_SUFFIX = "-completed";
+ implements RecoverableAggregationRepository, OptimisticLockingAggregationRepository {
+ private static final Logger LOG = LoggerFactory.getLogger(RedisAggregationRepository.class);
+ private static final String COMPLETED_SUFFIX = "-completed";
- private boolean optimistic;
- private boolean useRecovery = true;
- private Map<String, DefaultExchangeHolder> cache;
+ private boolean optimistic;
+ private boolean useRecovery = true;
+ private Map<String, DefaultExchangeHolder> cache;
private Map<String, DefaultExchangeHolder> persistedCache;
- private String endpoint;
- private String mapName;
- private String persistenceMapName;
- private RedissonClient redisson;
- private String deadLetterChannel;
- private long recoveryInterval = 5000;
- private int maximumRedeliveries = 3;
- private boolean allowSerializedHeaders;
-
- public RedisAggregationRepository(final String mapName, final String endpoint) {
- this.mapName = mapName;
- this.persistenceMapName = String.format("%s%s", mapName, COMPLETED_SUFFIX);
- this.optimistic = false;
- this.endpoint = endpoint;
- }
-
- public RedisAggregationRepository(final String mapName, final String persistenceMapName,
- final String endpoint) {
- this.mapName = mapName;
- this.persistenceMapName = persistenceMapName;
- this.optimistic = false;
- this.endpoint = endpoint;
- }
-
- public RedisAggregationRepository(final String mapName, final String endpoint, boolean optimistic) {
- this(mapName, endpoint);
- this.optimistic = optimistic;
- }
-
- public RedisAggregationRepository(final String mapName, final String persistenceMapName,
- final String endpoint, boolean optimistic) {
- this(mapName, persistenceMapName, endpoint);
- this.optimistic = optimistic;
- }
-
- @Override
- public Exchange add(CamelContext camelContext, String key, Exchange oldExchange, Exchange newExchange) throws OptimisticLockingException {
+ private String endpoint;
+ private String mapName;
+ private String persistenceMapName;
+ private RedissonClient redisson;
+ private boolean shutdownRedisson;
+ private String deadLetterChannel;
+ private long recoveryInterval = 5000;
+ private int maximumRedeliveries = 3;
+ private boolean allowSerializedHeaders;
+
+ public RedisAggregationRepository(final String mapName, final String endpoint) {
+ this.mapName = mapName;
+ this.persistenceMapName = String.format("%s%s", mapName, COMPLETED_SUFFIX);
+ this.optimistic = false;
+ this.endpoint = endpoint;
+ }
+
+ public RedisAggregationRepository(final String mapName, final String persistenceMapName,
+ final String endpoint) {
+ this.mapName = mapName;
+ this.persistenceMapName = persistenceMapName;
+ this.optimistic = false;
+ this.endpoint = endpoint;
+ }
+
+ public RedisAggregationRepository(final String mapName, final String endpoint, boolean optimistic) {
+ this(mapName, endpoint);
+ this.optimistic = optimistic;
+ }
+
+ public RedisAggregationRepository(final String mapName, final String persistenceMapName,
+ final String endpoint, boolean optimistic) {
+ this(mapName, persistenceMapName, endpoint);
+ this.optimistic = optimistic;
+ }
+
+ @Override
+ public Exchange add(CamelContext camelContext, String key, Exchange oldExchange, Exchange newExchange)
+ throws OptimisticLockingException {
if (!optimistic) {
throw new UnsupportedOperationException();
}
@@ -104,15 +99,17 @@ public class RedisAggregationRepository extends ServiceSupport
final DefaultExchangeHolder misbehaviorHolder = cache.putIfAbsent(key, holder);
if (misbehaviorHolder != null) {
Exchange misbehaviorEx = unmarshallExchange(camelContext, misbehaviorHolder);
- LOG.error("Optimistic locking failed for exchange with key {}: IMap#putIfAbsend returned Exchange with ID {}, while it's expected no exchanges to be returned",
+ LOG.warn(
+ "Optimistic locking failed for exchange with key {}: IMap#putIfAbsend returned Exchange with ID {}, while it's expected no exchanges to be returned",
key, misbehaviorEx != null ? misbehaviorEx.getExchangeId() : "<null>");
- throw new OptimisticLockingException();
+ throw new OptimisticLockingException();
}
} else {
DefaultExchangeHolder oldHolder = DefaultExchangeHolder.marshal(oldExchange, true, allowSerializedHeaders);
DefaultExchangeHolder newHolder = DefaultExchangeHolder.marshal(newExchange, true, allowSerializedHeaders);
if (!cache.replace(key, oldHolder, newHolder)) {
- LOG.error("Optimistic locking failed for exchange with key {}: IMap#replace returned no Exchanges, while it's expected to replace one",
+ LOG.warn(
+ "Optimistic locking failed for exchange with key {}: IMap#replace returned no Exchanges, while it's expected to replace one",
key);
throw new OptimisticLockingException();
}
@@ -121,14 +118,14 @@ public class RedisAggregationRepository extends ServiceSupport
return oldExchange;
}
- @Override
- public Exchange add(CamelContext camelContext, String key, Exchange exchange) {
- if (optimistic) {
- throw new UnsupportedOperationException();
- }
- LOG.trace("Adding an Exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(), key);
- RLock lock = redisson.getLock("aggregationLock");
- try {
+ @Override
+ public Exchange add(CamelContext camelContext, String key, Exchange exchange) {
+ if (optimistic) {
+ throw new UnsupportedOperationException();
+ }
+ LOG.trace("Adding an Exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(), key);
+ RLock lock = redisson.getLock("aggregationLock");
+ try {
lock.lock();
DefaultExchangeHolder newHolder = DefaultExchangeHolder.marshal(exchange, true, allowSerializedHeaders);
DefaultExchangeHolder oldHolder = cache.put(key, newHolder);
@@ -137,9 +134,9 @@ public class RedisAggregationRepository extends ServiceSupport
LOG.trace("Added an Exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(), key);
lock.unlock();
}
- }
+ }
- @Override
+ @Override
public Set<String> scan(CamelContext camelContext) {
if (useRecovery) {
LOG.trace("Scanning for exchanges to recover in {} context", camelContext.getName());
@@ -147,7 +144,8 @@ public class RedisAggregationRepository extends ServiceSupport
LOG.trace("Found {} keys for exchanges to recover in {} context", scanned.size(), camelContext.getName());
return scanned;
} else {
- LOG.warn("What for to run recovery scans in {} context while repository {} is running in non-recoverable aggregation repository mode?!",
+ LOG.warn(
+ "What for to run recovery scans in {} context while repository {} is running in non-recoverable aggregation repository mode?!",
camelContext.getName(), mapName);
return Collections.emptySet();
}
@@ -159,6 +157,46 @@ public class RedisAggregationRepository extends ServiceSupport
return useRecovery ? unmarshallExchange(camelContext, persistedCache.get(exchangeId)) : null;
}
+ public boolean isOptimistic() {
+ return optimistic;
+ }
+
+ public void setOptimistic(boolean optimistic) {
+ this.optimistic = optimistic;
+ }
+
+ public String getEndpoint() {
+ return endpoint;
+ }
+
+ public void setEndpoint(String endpoint) {
+ this.endpoint = endpoint;
+ }
+
+ public String getMapName() {
+ return mapName;
+ }
+
+ public void setMapName(String mapName) {
+ this.mapName = mapName;
+ }
+
+ public String getPersistenceMapName() {
+ return persistenceMapName;
+ }
+
+ public void setPersistenceMapName(String persistenceMapName) {
+ this.persistenceMapName = persistenceMapName;
+ }
+
+ public RedissonClient getRedisson() {
+ return redisson;
+ }
+
+ public void setRedisson(RedissonClient redisson) {
+ this.redisson = redisson;
+ }
+
@Override
public void setRecoveryInterval(long interval, TimeUnit timeUnit) {
this.recoveryInterval = timeUnit.toMillis(interval);
@@ -208,7 +246,7 @@ public class RedisAggregationRepository extends ServiceSupport
public Exchange get(CamelContext camelContext, String key) {
return unmarshallExchange(camelContext, cache.get(key));
}
-
+
/**
* Checks if the key in question is in the repository.
*
@@ -221,7 +259,7 @@ public class RedisAggregationRepository extends ServiceSupport
return false;
}
}
-
+
public boolean isAllowSerializedHeaders() {
return allowSerializedHeaders;
}
@@ -236,7 +274,8 @@ public class RedisAggregationRepository extends ServiceSupport
if (optimistic) {
LOG.trace("Removing an exchange with ID {} for key {} in an optimistic manner.", exchange.getExchangeId(), key);
if (!cache.remove(key, holder)) {
- LOG.error("Optimistic locking failed for exchange with key {}: IMap#remove removed no Exchanges, while it's expected to remove one.",
+ LOG.warn(
+ "Optimistic locking failed for exchange with key {}: IMap#remove removed no Exchanges, while it's expected to remove one.",
key);
throw new OptimisticLockingException();
}
@@ -255,8 +294,8 @@ public class RedisAggregationRepository extends ServiceSupport
RTransaction transaction = redisson.createTransaction(tOpts);
try {
- RMap<String, DefaultExchangeHolder> tCache = transaction.getMap(mapName);
- RMap<String, DefaultExchangeHolder> tPersistentCache = transaction.getMap(persistenceMapName);
+ RMap<String, DefaultExchangeHolder> tCache = transaction.getMap(mapName);
+ RMap<String, DefaultExchangeHolder> tPersistentCache = transaction.getMap(persistenceMapName);
DefaultExchangeHolder removedHolder = tCache.remove(key);
LOG.trace("Putting an exchange with ID {} for key {} into a recoverable storage in a thread-safe manner.",
@@ -264,14 +303,16 @@ public class RedisAggregationRepository extends ServiceSupport
tPersistentCache.put(exchange.getExchangeId(), removedHolder);
transaction.commit();
- LOG.trace("Removed an exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(), key);
+ LOG.trace("Removed an exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(),
+ key);
LOG.trace("Put an exchange with ID {} for key {} into a recoverable storage in a thread-safe manner.",
exchange.getExchangeId(), key);
} catch (Throwable throwable) {
- transaction.rollback();
+ transaction.rollback();
- final String msg = String.format("Transaction was rolled back for remove operation with a key %s and an Exchange ID %s.",
- key, exchange.getExchangeId());
+ final String msg = String.format(
+ "Transaction was rolled back for remove operation with a key %s and an Exchange ID %s.",
+ key, exchange.getExchangeId());
LOG.warn(msg, throwable);
throw new RuntimeException(msg, throwable);
}
@@ -294,42 +335,41 @@ public class RedisAggregationRepository extends ServiceSupport
return Collections.unmodifiableSet(cache.keySet());
}
- /**
- * @return Persistent repository {@link IMap} name;
- */
public String getPersistentRepositoryName() {
return persistenceMapName;
}
-
- @PostConstruct
- void init() throws Exception {
- if (maximumRedeliveries < 0) {
- throw new IllegalArgumentException("Maximum redelivery retries must be zero or a positive integer.");
- }
- if (recoveryInterval < 0) {
- throw new IllegalArgumentException("Recovery interval must be zero or a positive integer.");
- }
-
- }
-
- @Override
- protected void doStart() throws Exception {
- StringHelper.notEmpty(mapName, "repositoryName");
- Config config = new Config();
- config.useSingleServer().setAddress(String.format("redis://%s",endpoint));
- redisson = Redisson.create(config);
- cache = redisson.getMap(mapName);
- if (useRecovery) {
+
+ @Override
+ protected void doInit() throws Exception {
+ StringHelper.notEmpty(mapName, "repositoryName");
+ if (maximumRedeliveries < 0) {
+ throw new IllegalArgumentException("Maximum redelivery retries must be zero or a positive integer.");
+ }
+ if (recoveryInterval < 0) {
+ throw new IllegalArgumentException("Recovery interval must be zero or a positive integer.");
+ }
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ if (redisson == null) {
+ Config config = new Config();
+ config.useSingleServer().setAddress(String.format("redis://%s", endpoint));
+ redisson = Redisson.create(config);
+ shutdownRedisson = true;
+ }
+ cache = redisson.getMap(mapName);
+ if (useRecovery) {
persistedCache = redisson.getMap(persistenceMapName);
}
- }
-
- @Override
- protected void doStop() throws Exception {
- if(redisson!=null){
- redisson.shutdown();
- }
- }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (redisson != null && shutdownRedisson) {
+ redisson.shutdown();
+ }
+ }
protected Exchange unmarshallExchange(CamelContext camelContext, DefaultExchangeHolder holder) {
Exchange exchange = null;
diff --git a/components/camel-redis/src/test/AggregateRedisTest.java b/components/camel-redis/src/test/AggregateRedisTest.java
deleted file mode 100644
index 9867708..0000000
--- a/components/camel-redis/src/test/AggregateRedisTest.java
+++ /dev/null
@@ -1,51 +0,0 @@
-package org.apache.camel.component.redis;
-
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.junit4.CamelTestSupport;
-import org.apache.camel.component.redis.processor.aggregate.RedisAggregationRepository;
-import org.junit.Test;
-
-/**
- * The ABC example for using the Aggregator EIP.
- * <p/>
- * This example have 4 messages send to the aggregator, by which one
- * message is published which contains the aggregation of message 1,2 and 4
- * as they use the same correlation key.
- * <p/>
- * See the class {@link camelinaction.MyAggregationStrategy} for how the messages
- * are actually aggregated together.
- *
- * @see MyAggregationStrategy
- */
-public class AggregateRedisTest extends CamelTestSupport {
-
- private String endpoint = System.getProperty("endpoint"); //ip:port
-
- @Test
- public void testABC() throws Exception {
- MockEndpoint mock = getMockEndpoint("mock:result");
- mock.expectedBodiesReceived("ABC");
- template.sendBodyAndHeader("direct:start", "A", "myId", 1);
- template.sendBodyAndHeader("direct:start", "B", "myId", 1);
- template.sendBodyAndHeader("direct:start", "F", "myId", 2);
- template.sendBodyAndHeader("direct:start", "C", "myId", 1);
- assertMockEndpointsSatisfied();
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("direct:start")
- .log("Sending ${body} with correlation key ${header.myId}")
- .aggregate(header("myId"), new MyAggregationStrategy())
- .aggregationRepository(new RedisAggregationRepository("aggregation", endpoint))
- .completionSize(3)
- .log("Sending out ${body}")
- .to("mock:result");
- }
- };
- }
-}
\ No newline at end of file
diff --git a/components/camel-redis/src/test/java/org/apache/camel/component/redis/processor/aggregate/AggregateRedisTest.java b/components/camel-redis/src/test/java/org/apache/camel/component/redis/processor/aggregate/AggregateRedisTest.java
new file mode 100644
index 0000000..5629c8d
--- /dev/null
+++ b/components/camel-redis/src/test/java/org/apache/camel/component/redis/processor/aggregate/AggregateRedisTest.java
@@ -0,0 +1,67 @@
+/*
+ * 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.redis.processor.aggregate;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The ABC example for using the Aggregator EIP.
+ * <p/>
+ * This example have 4 messages send to the aggregator, by which one message is published which contains the aggregation
+ * of message 1,2 and 4 as they use the same correlation key.
+ * <p/>
+ */
+@Disabled("Requires manually testing")
+public class AggregateRedisTest extends CamelTestSupport {
+
+ // TODO: use docker test-containers for testing
+
+ private String endpoint = System.getProperty("endpoint"); //ip:port
+
+ @Test
+ public void testABC() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("ABC");
+
+ template.sendBodyAndHeader("direct:start", "A", "myId", 1);
+ template.sendBodyAndHeader("direct:start", "B", "myId", 1);
+ template.sendBodyAndHeader("direct:start", "F", "myId", 2);
+ template.sendBodyAndHeader("direct:start", "C", "myId", 1);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:start")
+ .log("Sending ${body} with correlation key ${header.myId}")
+ .aggregate(header("myId"), new MyAggregationStrategy())
+ .aggregationRepository(new RedisAggregationRepository("aggregation", endpoint))
+ .completionSize(3)
+ .log("Sending out ${body}")
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git a/components/camel-redis/src/test/MyAggregationStrategy.java b/components/camel-redis/src/test/java/org/apache/camel/component/redis/processor/aggregate/MyAggregationStrategy.java
similarity index 68%
rename from components/camel-redis/src/test/MyAggregationStrategy.java
rename to components/camel-redis/src/test/java/org/apache/camel/component/redis/processor/aggregate/MyAggregationStrategy.java
index ed650e2..4de8c75 100644
--- a/components/camel-redis/src/test/MyAggregationStrategy.java
+++ b/components/camel-redis/src/test/java/org/apache/camel/component/redis/processor/aggregate/MyAggregationStrategy.java
@@ -1,20 +1,21 @@
+package org.apache.camel.component.redis.processor.aggregate;
+
+import org.apache.camel.AggregationStrategy;
import org.apache.camel.Exchange;
-import org.apache.camel.processor.aggregate.AggregationStrategy;
/**
- * This is the aggregation strategy which is java code for <i>aggregating</i>
- * incoming messages with the existing aggregated message. In other words
- * you use this strategy to <i>merge</i> the messages together.
+ * This is the aggregation strategy which is java code for <i>aggregating</i> incoming messages with the existing
+ * aggregated message. In other words you use this strategy to <i>merge</i> the messages together.
*/
public class MyAggregationStrategy implements AggregationStrategy {
/**
* Aggregates the messages.
*
- * @param oldExchange the existing aggregated message. Is <tt>null</tt> the
- * very first time as there are no existing message.
- * @param newExchange the incoming message. This is never <tt>null</tt>.
- * @return the aggregated message.
+ * @param oldExchange the existing aggregated message. Is <tt>null</tt> the very first time as there are no
+ * existing message.
+ * @param newExchange the incoming message. This is never <tt>null</tt>.
+ * @return the aggregated message.
*/
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
// the first time there are no existing message and therefore
@@ -40,5 +41,5 @@ public class MyAggregationStrategy implements AggregationStrategy {
// and return it
return oldExchange;
}
-
-}
\ No newline at end of file
+
+}
diff --git a/components/camel-redis/src/test/resources/log4j2.properties b/components/camel-redis/src/test/resources/log4j2.properties
new file mode 100644
index 0000000..8728a5d
--- /dev/null
+++ b/components/camel-redis/src/test/resources/log4j2.properties
@@ -0,0 +1,28 @@
+## ---------------------------------------------------------------------------
+## 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-redis-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
diff --git a/docs/components/modules/others/nav.adoc b/docs/components/modules/others/nav.adoc
index 3a4fed6..514144c 100644
--- a/docs/components/modules/others/nav.adoc
+++ b/docs/components/modules/others/nav.adoc
@@ -22,6 +22,7 @@
** xref:platform-http-vertx.adoc[Platform Http Vertx]
** xref:reactive-executor-vertx.adoc[Reactive Executor Vert.x]
** xref:reactor.adoc[Reactor]
+** xref:redis.adoc[Redis]
** xref:resilience4j.adoc[Resilience4j]
** xref:ribbon.adoc[Ribbon]
** xref:rxjava.adoc[RxJava]
diff --git a/docs/components/modules/others/pages/redis.adoc b/docs/components/modules/others/pages/redis.adoc
new file mode 100644
index 0000000..a24c476
--- /dev/null
+++ b/docs/components/modules/others/pages/redis.adoc
@@ -0,0 +1,13 @@
+[[redis-component]]
+= Redis Component
+//THIS FILE IS COPIED: EDIT THE SOURCE FILE:
+:page-source: components/camel-redis/src/main/docs/redis.adoc
+:docTitle: Redis
+:artifactId: camel-redis
+:description: Aggregation repository using Redis as datastore
+:since: 3.5
+:supportLevel: Preview
+
+*Since Camel {since}*
+
+The Redis component provides an `AggregationStrategy` to use Redis as the backend datastore.
diff --git a/parent/pom.xml b/parent/pom.xml
index 50ab1ff..444086d 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -2066,7 +2066,7 @@
<artifactId>camel-reactor</artifactId>
<version>${project.version}</version>
</dependency>
- <dependency>
+ <dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-redis</artifactId>
<version>${project.version}</version>
[camel] 01/02: Added camel-redis component and
RedisAggregationRepository using Redisson
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 0d754837c6d531b7f350ca1d789685e6a4da6522
Author: malogan82 <mr...@gmail.com>
AuthorDate: Sat Jul 11 22:26:56 2020 +0200
Added camel-redis component and RedisAggregationRepository using Redisson
Added camel-redis component and RedisAggregationRepository using Redisson
Added test class and fixed some bugs
Modified test class
Modified test class
Modified RedissonAggregationRepository
---
bom/camel-bom/pom.xml | 5 +
camel-dependencies/pom.xml | 1 +
components/camel-redis/pom.xml | 64 ++++
.../aggregate/RedisAggregationRepository.java | 342 +++++++++++++++++++++
.../camel-redis/src/test/AggregateRedisTest.java | 51 +++
.../src/test/MyAggregationStrategy.java | 44 +++
components/pom.xml | 1 +
parent/pom.xml | 6 +
8 files changed, 514 insertions(+)
diff --git a/bom/camel-bom/pom.xml b/bom/camel-bom/pom.xml
index 0118577..21320c9 100644
--- a/bom/camel-bom/pom.xml
+++ b/bom/camel-bom/pom.xml
@@ -1460,6 +1460,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-redis</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-ref</artifactId>
<version>${project.version}</version>
</dependency>
diff --git a/camel-dependencies/pom.xml b/camel-dependencies/pom.xml
index 92859c5..e66d598 100644
--- a/camel-dependencies/pom.xml
+++ b/camel-dependencies/pom.xml
@@ -489,6 +489,7 @@
<rdf4j-rio-version>2.4.4</rdf4j-rio-version>
<reactive-streams-version>1.0.3</reactive-streams-version>
<reactor-version>3.2.16.RELEASE</reactor-version>
+ <redisson-version>3.13.2</redisson-version>
<rescu-version>2.0.2</rescu-version>
<resilience4j-version>1.5.0</resilience4j-version>
<rest-assured-version>4.3.1</rest-assured-version>
diff --git a/components/camel-redis/pom.xml b/components/camel-redis/pom.xml
new file mode 100644
index 0000000..700269e
--- /dev/null
+++ b/components/camel-redis/pom.xml
@@ -0,0 +1,64 @@
+<?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>3.5.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-redis</artifactId>
+ <packaging>jar</packaging>
+ <name>Camel :: Redis</name>
+ <description>Camel Redis support</description>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.redisson</groupId>
+ <artifactId>redisson</artifactId>
+ <version>${redisson-version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-spring-junit5</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <endpoint>${endpoint}</endpoint>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/components/camel-redis/src/main/java/org/apache/camel/component/redis/processor/aggregate/RedisAggregationRepository.java b/components/camel-redis/src/main/java/org/apache/camel/component/redis/processor/aggregate/RedisAggregationRepository.java
new file mode 100644
index 0000000..4b5ab34
--- /dev/null
+++ b/components/camel-redis/src/main/java/org/apache/camel/component/redis/processor/aggregate/RedisAggregationRepository.java
@@ -0,0 +1,342 @@
+/*
+ * 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.redis.processor.aggregate;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.spi.OptimisticLockingAggregationRepository;
+import org.apache.camel.spi.RecoverableAggregationRepository;
+import org.apache.camel.spi.OptimisticLockingAggregationRepository.OptimisticLockingException;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.DefaultExchangeHolder;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.util.StringHelper;
+import org.redisson.Redisson;
+import org.redisson.api.RLock;
+import org.redisson.api.RMap;
+import org.redisson.api.RTransaction;
+import org.redisson.api.RedissonClient;
+import org.redisson.api.TransactionOptions;
+import org.redisson.config.Config;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RedisAggregationRepository extends ServiceSupport
+ implements RecoverableAggregationRepository, OptimisticLockingAggregationRepository {
+ private static final Logger LOG = LoggerFactory.getLogger(RedisAggregationRepository.class);
+ private static final String COMPLETED_SUFFIX = "-completed";
+
+ private boolean optimistic;
+ private boolean useRecovery = true;
+ private Map<String, DefaultExchangeHolder> cache;
+ private Map<String, DefaultExchangeHolder> persistedCache;
+ private String endpoint;
+ private String mapName;
+ private String persistenceMapName;
+ private RedissonClient redisson;
+ private String deadLetterChannel;
+ private long recoveryInterval = 5000;
+ private int maximumRedeliveries = 3;
+ private boolean allowSerializedHeaders;
+
+ public RedisAggregationRepository(final String mapName, final String endpoint) {
+ this.mapName = mapName;
+ this.persistenceMapName = String.format("%s%s", mapName, COMPLETED_SUFFIX);
+ this.optimistic = false;
+ this.endpoint = endpoint;
+ }
+
+ public RedisAggregationRepository(final String mapName, final String persistenceMapName,
+ final String endpoint) {
+ this.mapName = mapName;
+ this.persistenceMapName = persistenceMapName;
+ this.optimistic = false;
+ this.endpoint = endpoint;
+ }
+
+ public RedisAggregationRepository(final String mapName, final String endpoint, boolean optimistic) {
+ this(mapName, endpoint);
+ this.optimistic = optimistic;
+ }
+
+ public RedisAggregationRepository(final String mapName, final String persistenceMapName,
+ final String endpoint, boolean optimistic) {
+ this(mapName, persistenceMapName, endpoint);
+ this.optimistic = optimistic;
+ }
+
+ @Override
+ public Exchange add(CamelContext camelContext, String key, Exchange oldExchange, Exchange newExchange) throws OptimisticLockingException {
+ if (!optimistic) {
+ throw new UnsupportedOperationException();
+ }
+ LOG.trace("Adding an Exchange with ID {} for key {} in an optimistic manner.", newExchange.getExchangeId(), key);
+ if (oldExchange == null) {
+ DefaultExchangeHolder holder = DefaultExchangeHolder.marshal(newExchange, true, allowSerializedHeaders);
+ final DefaultExchangeHolder misbehaviorHolder = cache.putIfAbsent(key, holder);
+ if (misbehaviorHolder != null) {
+ Exchange misbehaviorEx = unmarshallExchange(camelContext, misbehaviorHolder);
+ LOG.error("Optimistic locking failed for exchange with key {}: IMap#putIfAbsend returned Exchange with ID {}, while it's expected no exchanges to be returned",
+ key, misbehaviorEx != null ? misbehaviorEx.getExchangeId() : "<null>");
+ throw new OptimisticLockingException();
+ }
+ } else {
+ DefaultExchangeHolder oldHolder = DefaultExchangeHolder.marshal(oldExchange, true, allowSerializedHeaders);
+ DefaultExchangeHolder newHolder = DefaultExchangeHolder.marshal(newExchange, true, allowSerializedHeaders);
+ if (!cache.replace(key, oldHolder, newHolder)) {
+ LOG.error("Optimistic locking failed for exchange with key {}: IMap#replace returned no Exchanges, while it's expected to replace one",
+ key);
+ throw new OptimisticLockingException();
+ }
+ }
+ LOG.trace("Added an Exchange with ID {} for key {} in optimistic manner.", newExchange.getExchangeId(), key);
+ return oldExchange;
+ }
+
+ @Override
+ public Exchange add(CamelContext camelContext, String key, Exchange exchange) {
+ if (optimistic) {
+ throw new UnsupportedOperationException();
+ }
+ LOG.trace("Adding an Exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(), key);
+ RLock lock = redisson.getLock("aggregationLock");
+ try {
+ lock.lock();
+ DefaultExchangeHolder newHolder = DefaultExchangeHolder.marshal(exchange, true, allowSerializedHeaders);
+ DefaultExchangeHolder oldHolder = cache.put(key, newHolder);
+ return unmarshallExchange(camelContext, oldHolder);
+ } finally {
+ LOG.trace("Added an Exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(), key);
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public Set<String> scan(CamelContext camelContext) {
+ if (useRecovery) {
+ LOG.trace("Scanning for exchanges to recover in {} context", camelContext.getName());
+ Set<String> scanned = Collections.unmodifiableSet(persistedCache.keySet());
+ LOG.trace("Found {} keys for exchanges to recover in {} context", scanned.size(), camelContext.getName());
+ return scanned;
+ } else {
+ LOG.warn("What for to run recovery scans in {} context while repository {} is running in non-recoverable aggregation repository mode?!",
+ camelContext.getName(), mapName);
+ return Collections.emptySet();
+ }
+ }
+
+ @Override
+ public Exchange recover(CamelContext camelContext, String exchangeId) {
+ LOG.trace("Recovering an Exchange with ID {}.", exchangeId);
+ return useRecovery ? unmarshallExchange(camelContext, persistedCache.get(exchangeId)) : null;
+ }
+
+ @Override
+ public void setRecoveryInterval(long interval, TimeUnit timeUnit) {
+ this.recoveryInterval = timeUnit.toMillis(interval);
+ }
+
+ @Override
+ public void setRecoveryInterval(long interval) {
+ this.recoveryInterval = interval;
+ }
+
+ @Override
+ public long getRecoveryIntervalInMillis() {
+ return recoveryInterval;
+ }
+
+ @Override
+ public void setUseRecovery(boolean useRecovery) {
+ this.useRecovery = useRecovery;
+ }
+
+ @Override
+ public boolean isUseRecovery() {
+ return useRecovery;
+ }
+
+ @Override
+ public void setDeadLetterUri(String deadLetterUri) {
+ this.deadLetterChannel = deadLetterUri;
+ }
+
+ @Override
+ public String getDeadLetterUri() {
+ return deadLetterChannel;
+ }
+
+ @Override
+ public void setMaximumRedeliveries(int maximumRedeliveries) {
+ this.maximumRedeliveries = maximumRedeliveries;
+ }
+
+ @Override
+ public int getMaximumRedeliveries() {
+ return maximumRedeliveries;
+ }
+
+ @Override
+ public Exchange get(CamelContext camelContext, String key) {
+ return unmarshallExchange(camelContext, cache.get(key));
+ }
+
+ /**
+ * Checks if the key in question is in the repository.
+ *
+ * @param key Object - key in question
+ */
+ public boolean containsKey(Object key) {
+ if (cache != null) {
+ return cache.containsKey(key);
+ } else {
+ return false;
+ }
+ }
+
+ public boolean isAllowSerializedHeaders() {
+ return allowSerializedHeaders;
+ }
+
+ public void setAllowSerializedHeaders(boolean allowSerializedHeaders) {
+ this.allowSerializedHeaders = allowSerializedHeaders;
+ }
+
+ @Override
+ public void remove(CamelContext camelContext, String key, Exchange exchange) {
+ DefaultExchangeHolder holder = DefaultExchangeHolder.marshal(exchange, true, allowSerializedHeaders);
+ if (optimistic) {
+ LOG.trace("Removing an exchange with ID {} for key {} in an optimistic manner.", exchange.getExchangeId(), key);
+ if (!cache.remove(key, holder)) {
+ LOG.error("Optimistic locking failed for exchange with key {}: IMap#remove removed no Exchanges, while it's expected to remove one.",
+ key);
+ throw new OptimisticLockingException();
+ }
+ LOG.trace("Removed an exchange with ID {} for key {} in an optimistic manner.", exchange.getExchangeId(), key);
+ if (useRecovery) {
+ LOG.trace("Putting an exchange with ID {} for key {} into a recoverable storage in an optimistic manner.",
+ exchange.getExchangeId(), key);
+ persistedCache.put(exchange.getExchangeId(), holder);
+ LOG.trace("Put an exchange with ID {} for key {} into a recoverable storage in an optimistic manner.",
+ exchange.getExchangeId(), key);
+ }
+ } else {
+ if (useRecovery) {
+ LOG.trace("Removing an exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(), key);
+ TransactionOptions tOpts = TransactionOptions.defaults();
+ RTransaction transaction = redisson.createTransaction(tOpts);
+
+ try {
+ RMap<String, DefaultExchangeHolder> tCache = transaction.getMap(mapName);
+ RMap<String, DefaultExchangeHolder> tPersistentCache = transaction.getMap(persistenceMapName);
+
+ DefaultExchangeHolder removedHolder = tCache.remove(key);
+ LOG.trace("Putting an exchange with ID {} for key {} into a recoverable storage in a thread-safe manner.",
+ exchange.getExchangeId(), key);
+ tPersistentCache.put(exchange.getExchangeId(), removedHolder);
+
+ transaction.commit();
+ LOG.trace("Removed an exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(), key);
+ LOG.trace("Put an exchange with ID {} for key {} into a recoverable storage in a thread-safe manner.",
+ exchange.getExchangeId(), key);
+ } catch (Throwable throwable) {
+ transaction.rollback();
+
+ final String msg = String.format("Transaction was rolled back for remove operation with a key %s and an Exchange ID %s.",
+ key, exchange.getExchangeId());
+ LOG.warn(msg, throwable);
+ throw new RuntimeException(msg, throwable);
+ }
+ } else {
+ cache.remove(key);
+ }
+ }
+ }
+
+ @Override
+ public void confirm(CamelContext camelContext, String exchangeId) {
+ LOG.trace("Confirming an exchange with ID {}.", exchangeId);
+ if (useRecovery) {
+ persistedCache.remove(exchangeId);
+ }
+ }
+
+ @Override
+ public Set<String> getKeys() {
+ return Collections.unmodifiableSet(cache.keySet());
+ }
+
+ /**
+ * @return Persistent repository {@link IMap} name;
+ */
+ public String getPersistentRepositoryName() {
+ return persistenceMapName;
+ }
+
+ @PostConstruct
+ void init() throws Exception {
+ if (maximumRedeliveries < 0) {
+ throw new IllegalArgumentException("Maximum redelivery retries must be zero or a positive integer.");
+ }
+ if (recoveryInterval < 0) {
+ throw new IllegalArgumentException("Recovery interval must be zero or a positive integer.");
+ }
+
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ StringHelper.notEmpty(mapName, "repositoryName");
+ Config config = new Config();
+ config.useSingleServer().setAddress(String.format("redis://%s",endpoint));
+ redisson = Redisson.create(config);
+ cache = redisson.getMap(mapName);
+ if (useRecovery) {
+ persistedCache = redisson.getMap(persistenceMapName);
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if(redisson!=null){
+ redisson.shutdown();
+ }
+ }
+
+ protected Exchange unmarshallExchange(CamelContext camelContext, DefaultExchangeHolder holder) {
+ Exchange exchange = null;
+ if (holder != null) {
+ exchange = new DefaultExchange(camelContext);
+ DefaultExchangeHolder.unmarshal(exchange, holder);
+ }
+ return exchange;
+ }
+}
diff --git a/components/camel-redis/src/test/AggregateRedisTest.java b/components/camel-redis/src/test/AggregateRedisTest.java
new file mode 100644
index 0000000..9867708
--- /dev/null
+++ b/components/camel-redis/src/test/AggregateRedisTest.java
@@ -0,0 +1,51 @@
+package org.apache.camel.component.redis;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.apache.camel.component.redis.processor.aggregate.RedisAggregationRepository;
+import org.junit.Test;
+
+/**
+ * The ABC example for using the Aggregator EIP.
+ * <p/>
+ * This example have 4 messages send to the aggregator, by which one
+ * message is published which contains the aggregation of message 1,2 and 4
+ * as they use the same correlation key.
+ * <p/>
+ * See the class {@link camelinaction.MyAggregationStrategy} for how the messages
+ * are actually aggregated together.
+ *
+ * @see MyAggregationStrategy
+ */
+public class AggregateRedisTest extends CamelTestSupport {
+
+ private String endpoint = System.getProperty("endpoint"); //ip:port
+
+ @Test
+ public void testABC() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("ABC");
+ template.sendBodyAndHeader("direct:start", "A", "myId", 1);
+ template.sendBodyAndHeader("direct:start", "B", "myId", 1);
+ template.sendBodyAndHeader("direct:start", "F", "myId", 2);
+ template.sendBodyAndHeader("direct:start", "C", "myId", 1);
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:start")
+ .log("Sending ${body} with correlation key ${header.myId}")
+ .aggregate(header("myId"), new MyAggregationStrategy())
+ .aggregationRepository(new RedisAggregationRepository("aggregation", endpoint))
+ .completionSize(3)
+ .log("Sending out ${body}")
+ .to("mock:result");
+ }
+ };
+ }
+}
\ No newline at end of file
diff --git a/components/camel-redis/src/test/MyAggregationStrategy.java b/components/camel-redis/src/test/MyAggregationStrategy.java
new file mode 100644
index 0000000..ed650e2
--- /dev/null
+++ b/components/camel-redis/src/test/MyAggregationStrategy.java
@@ -0,0 +1,44 @@
+import org.apache.camel.Exchange;
+import org.apache.camel.processor.aggregate.AggregationStrategy;
+
+/**
+ * This is the aggregation strategy which is java code for <i>aggregating</i>
+ * incoming messages with the existing aggregated message. In other words
+ * you use this strategy to <i>merge</i> the messages together.
+ */
+public class MyAggregationStrategy implements AggregationStrategy {
+
+ /**
+ * Aggregates the messages.
+ *
+ * @param oldExchange the existing aggregated message. Is <tt>null</tt> the
+ * very first time as there are no existing message.
+ * @param newExchange the incoming message. This is never <tt>null</tt>.
+ * @return the aggregated message.
+ */
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ // the first time there are no existing message and therefore
+ // the oldExchange is null. In these cases we just return
+ // the newExchange
+ if (oldExchange == null) {
+ return newExchange;
+ }
+
+ // now we have both an existing message (oldExchange)
+ // and a incoming message (newExchange)
+ // we want to merge together.
+
+ // in this example we add their bodies
+ String oldBody = oldExchange.getIn().getBody(String.class).trim();
+ String newBody = newExchange.getIn().getBody(String.class).trim();
+
+ // the body should be the two bodies added together
+ String body = oldBody + newBody;
+
+ // update the existing message with the added body
+ oldExchange.getIn().setBody(body);
+ // and return it
+ return oldExchange;
+ }
+
+}
\ No newline at end of file
diff --git a/components/pom.xml b/components/pom.xml
index b438cb5..5d89fe7 100644
--- a/components/pom.xml
+++ b/components/pom.xml
@@ -315,6 +315,7 @@
<module>camel-reactive-executor-vertx</module>
<module>camel-reactive-streams</module>
<module>camel-reactor</module>
+ <module>camel-redis</module>
<module>camel-resilience4j</module>
<module>camel-resteasy</module>
<module>camel-rest-swagger</module>
diff --git a/parent/pom.xml b/parent/pom.xml
index 3fb163b..50ab1ff 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -477,6 +477,7 @@
<rdf4j-model-version>2.4.4</rdf4j-model-version>
<reactive-streams-version>1.0.3</reactive-streams-version>
<reactor-version>3.2.16.RELEASE</reactor-version>
+ <redisson-version>3.13.2</redisson-version>
<rescu-version>2.0.2</rescu-version>
<resilience4j-version>1.5.0</resilience4j-version>
<rest-assured-version>4.3.1</rest-assured-version>
@@ -2065,6 +2066,11 @@
<artifactId>camel-reactor</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-redis</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-ref</artifactId>