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 2022/04/21 05:14:41 UTC
[camel-spring-boot] branch main updated: CAMEL-17907: Add camel-infinispan-starter test (#514)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-spring-boot.git
The following commit(s) were added to refs/heads/main by this push:
new 16a8f77f888 CAMEL-17907: Add camel-infinispan-starter test (#514)
16a8f77f888 is described below
commit 16a8f77f8889cb4c70751a6f41897f341c0db092
Author: Federico Mariani <34...@users.noreply.github.com>
AuthorDate: Thu Apr 21 07:14:37 2022 +0200
CAMEL-17907: Add camel-infinispan-starter test (#514)
---
.../camel-infinispan-starter/pom.xml | 122 +++
.../infinispan/InfinispanConsumerTestSupport.java | 32 +
.../InfinispanIdempotentRepositoryTestSupport.java | 103 +++
.../infinispan/InfinispanProducerTestSupport.java | 870 +++++++++++++++++++++
.../InfinispanRemoteAggregationRepositoryIT.java | 97 +++
...panRemoteAggregationRepositoryOperationsIT.java | 246 ++++++
.../remote/InfinispanRemoteConfigurationIT.java | 120 +++
.../remote/InfinispanRemoteConsumerIT.java | 90 +++
.../remote/InfinispanRemoteCustomListenerIT.java | 114 +++
.../InfinispanRemoteIdempotentRepositoryIT.java | 94 +++
.../remote/InfinispanRemoteProducerIT.java | 118 +++
.../remote/InfinispanRemoteQueryConsumerIT.java | 157 ++++
.../remote/InfinispanRemoteQueryProducerIT.java | 184 +++++
.../remote/InfinispanRemoteQueryTestSupport.java | 80 ++
.../remote/InfinispanRemoteTestSupport.java | 154 ++++
.../InfinispanRemoteClusteredTestSupport.java | 74 ++
.../cluster/InfinispanRemoteClusteredViewIT.java | 95 +++
...nfinispanRemoteIdempotentRepositoryCamelIT.java | 54 ++
...finispanRemoteIdempotentRepositorySpringIT.java | 68 ++
...ispanRemoteIdempotentRepositoryTestSupport.java | 100 +++
.../resources/infinispan/client-mac.properties | 24 +
.../test/resources/infinispan/client.properties | 20 +
.../src/test/resources/log4j2.properties | 48 ++
...finispanRemoteIdempotentRepositoryCamelTest.xml | 29 +
...inispanRemoteIdempotentRepositorySpringTest.xml | 29 +
25 files changed, 3122 insertions(+)
diff --git a/components-starter/camel-infinispan-starter/pom.xml b/components-starter/camel-infinispan-starter/pom.xml
index d5e2af7dfdb..233ec143712 100644
--- a/components-starter/camel-infinispan-starter/pom.xml
+++ b/components-starter/camel-infinispan-starter/pom.xml
@@ -38,6 +38,66 @@
<artifactId>camel-infinispan</artifactId>
<version>${camel-version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-spring-junit5</artifactId>
+ <version>${camel-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel.springboot</groupId>
+ <artifactId>camel-master-starter</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-infinispan</artifactId>
+ <version>${camel-version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <!-- testing - infinispan -->
+ <dependency>
+ <groupId>org.infinispan</groupId>
+ <artifactId>infinispan-core</artifactId>
+ <version>${infinispan-version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.infinispan</groupId>
+ <artifactId>infinispan-commons-test</artifactId>
+ <version>${infinispan-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.infinispan</groupId>
+ <artifactId>infinispan-spring5-remote</artifactId>
+ <version>${infinispan-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.infinispan.protostream</groupId>
+ <artifactId>sample-domain-definition</artifactId>
+ <version>${protostream-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.infinispan.protostream</groupId>
+ <artifactId>sample-domain-implementation</artifactId>
+ <version>${protostream-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<!--START OF GENERATED CODE-->
<dependency>
<groupId>org.apache.camel.springboot</groupId>
@@ -45,4 +105,66 @@
</dependency>
<!--END OF GENERATED CODE-->
</dependencies>
+ <profiles>
+ <!-- activate integration test if the docker socket file is accessible -->
+ <profile>
+ <id>infinispan-integration-tests-docker-file</id>
+ <activation>
+ <file>
+ <exists>/var/run/docker.sock</exists>
+ </file>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <visibleassertions.silence>true</visibleassertions.silence>
+ </systemPropertyVariables>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <!-- activate integration test if the DOCKER_HOST env var is set -->
+ <profile>
+ <id>infinispan-integration-tests-docker-env</id>
+ <activation>
+ <property>
+ <name>env.DOCKER_HOST</name>
+ </property>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <visibleassertions.silence>true</visibleassertions.silence>
+ </systemPropertyVariables>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
</project>
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/InfinispanConsumerTestSupport.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/InfinispanConsumerTestSupport.java
new file mode 100644
index 00000000000..15017d5fa79
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/InfinispanConsumerTestSupport.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.infinispan;
+
+import org.apache.camel.component.mock.MockEndpoint;
+import org.infinispan.commons.api.BasicCache;
+
+public interface InfinispanConsumerTestSupport {
+ String KEY_ONE = "keyOne";
+ String VALUE_ONE = "valueOne";
+ String VALUE_TWO = "valueTwo";
+
+ BasicCache<Object, Object> getCache();
+
+ BasicCache<Object, Object> getCache(String name);
+
+ MockEndpoint getMockEndpoint(String id);
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/InfinispanIdempotentRepositoryTestSupport.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/InfinispanIdempotentRepositoryTestSupport.java
new file mode 100644
index 00000000000..57237f99f42
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/InfinispanIdempotentRepositoryTestSupport.java
@@ -0,0 +1,103 @@
+/*
+ * 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.infinispan;
+
+import java.util.UUID;
+import java.util.stream.IntStream;
+
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.IdempotentRepository;
+import org.infinispan.commons.api.BasicCache;
+import org.junit.jupiter.api.Test;
+
+import static org.jgroups.util.Util.assertFalse;
+import static org.jgroups.util.Util.assertTrue;
+
+public interface InfinispanIdempotentRepositoryTestSupport {
+ IdempotentRepository getIdempotentRepository();
+
+ BasicCache<Object, Object> getCache();
+
+ MockEndpoint getMockEndpoint(String id);
+
+ ProducerTemplate template();
+
+ @Test
+ default void addsNewKeysToCache() {
+ assertTrue(getIdempotentRepository().add("One"));
+ assertTrue(getIdempotentRepository().add("Two"));
+
+ assertTrue(getCache().containsKey("One"));
+ assertTrue(getCache().containsKey("Two"));
+ }
+
+ @Test
+ default void skipsAddingSecondTimeTheSameKey() {
+ assertTrue(getIdempotentRepository().add("One"));
+ assertFalse(getIdempotentRepository().add("One"));
+ }
+
+ @Test
+ default void containsPreviouslyAddedKey() {
+ assertFalse(getIdempotentRepository().contains("One"));
+
+ getIdempotentRepository().add("One");
+
+ assertTrue(getIdempotentRepository().contains("One"));
+ }
+
+ @Test
+ default void removesAnExistingKey() {
+ getIdempotentRepository().add("One");
+
+ assertTrue(getIdempotentRepository().remove("One"));
+
+ assertFalse(getIdempotentRepository().contains("One"));
+ }
+
+ @Test
+ default void doesntRemoveMissingKey() {
+ assertFalse(getIdempotentRepository().remove("One"));
+ }
+
+ @Test
+ default void clearCache() {
+ assertTrue(getIdempotentRepository().add("One"));
+ assertTrue(getIdempotentRepository().add("Two"));
+
+ assertTrue(getCache().containsKey("One"));
+ assertTrue(getCache().containsKey("Two"));
+
+ getIdempotentRepository().clear();
+
+ assertFalse(getCache().containsKey("One"));
+ assertFalse(getCache().containsKey("Two"));
+ }
+
+ @Test
+ default void producerQueryOperationWithoutQueryBuilder() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ final String messageId = UUID.randomUUID().toString();
+ IntStream.range(0, 10).forEach(
+ i -> template().sendBodyAndHeader("direct:start", "message-" + i, "MessageID", messageId));
+
+ mock.assertIsSatisfied();
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/InfinispanProducerTestSupport.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/InfinispanProducerTestSupport.java
new file mode 100644
index 00000000000..cffdeb8506a
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/InfinispanProducerTestSupport.java
@@ -0,0 +1,870 @@
+/*
+ * 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.infinispan;
+
+import java.security.SecureRandom;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.FluentProducerTemplate;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.infinispan.InfinispanConstants;
+import org.apache.camel.component.infinispan.InfinispanOperation;
+import org.apache.camel.util.CollectionHelper;
+import org.awaitility.Awaitility;
+import org.infinispan.commons.api.BasicCache;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public interface InfinispanProducerTestSupport {
+ String KEY_ONE = "keyOne";
+ String VALUE_ONE = "valueOne";
+ String KEY_TWO = "keyTwo";
+ String VALUE_TWO = "valueTwo";
+ String COMMAND_VALUE = "commandValue";
+ String COMMAND_KEY = "commandKey1";
+ long LIFESPAN_TIME = 300;
+ long LIFESPAN_FOR_MAX_IDLE = -1;
+ long MAX_IDLE_TIME = 500;
+
+ static void wait(long timout, Callable<Boolean> condition) {
+ Awaitility.await()
+ // wait at most the given timeout before failing
+ .atMost(timout, TimeUnit.MILLISECONDS)
+ // the condition to assert
+ .until(condition);
+ }
+
+ static void wait(long delay, long timout, Callable<Boolean> condition) {
+ long jitter = 50 + new SecureRandom().nextInt(50);
+
+ Awaitility.await()
+ // wait for the given delay (plus some jitter)
+ .pollDelay(delay + jitter, TimeUnit.MILLISECONDS)
+ // wait at most the given timeout before failing
+ .atMost(timout, TimeUnit.MILLISECONDS)
+ // the condition to assert
+ .until(condition);
+ }
+
+ BasicCache<Object, Object> getCache();
+
+ BasicCache<Object, Object> getCache(String name);
+
+ ProducerTemplate template();
+
+ FluentProducerTemplate fluentTemplate();
+
+ @Test
+ default void keyAndValueArePublishedWithDefaultOperation() {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .send();
+
+ Object value = getCache().get(KEY_ONE);
+ assertEquals(VALUE_ONE, value.toString());
+ }
+
+ @Test
+ default void cacheSizeTest() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+ getCache().put(KEY_TWO, VALUE_TWO);
+
+ Integer cacheSize = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.SIZE)
+ .request(Integer.class);
+
+ assertEquals(cacheSize, Integer.valueOf(2));
+ }
+
+ @Test
+ default void publishKeyAndValueByExplicitlySpecifyingTheOperation() {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUT)
+ .send();
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ }
+
+ @Test
+ default void publishKeyAndValueAsync() throws Exception {
+ assertTrue(getCache().isEmpty());
+
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTASYNC)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ }
+
+ @Test
+ default void publishKeyAndValueAsyncWithLifespan() throws Exception {
+ assertTrue(getCache().isEmpty());
+
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTASYNC)
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_TIME)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+
+ wait(LIFESPAN_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void publishKeyAndValueAsyncWithLifespanAndMaxIdle() throws Exception {
+ assertTrue(getCache().isEmpty());
+
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTASYNC)
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_FOR_MAX_IDLE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME, MAX_IDLE_TIME)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+
+ wait(MAX_IDLE_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void publishMapNormal() {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.MAP, CollectionHelper.mapOf(KEY_ONE, VALUE_ONE, KEY_TWO, VALUE_TWO))
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTALL)
+ .send();
+
+ assertEquals(2, getCache().size());
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ assertEquals(VALUE_TWO, getCache().get(KEY_TWO));
+ }
+
+ @Test
+ default void publishMapWithLifespan() throws Exception {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.MAP, CollectionHelper.mapOf(KEY_ONE, VALUE_ONE, KEY_TWO, VALUE_TWO))
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTALL)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_TIME)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .send();
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ assertEquals(VALUE_TWO, getCache().get(KEY_TWO));
+
+ wait(LIFESPAN_TIME, 5000, () -> !getCache().containsKey(KEY_ONE) && !getCache().containsKey(KEY_TWO));
+ }
+
+ @Test
+ default void publishMapWithLifespanAndMaxIdleTime() throws Exception {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.MAP, CollectionHelper.mapOf(KEY_ONE, VALUE_ONE, KEY_TWO, VALUE_TWO))
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTALL)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_FOR_MAX_IDLE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME, MAX_IDLE_TIME)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .send();
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ assertEquals(VALUE_TWO, getCache().get(KEY_TWO));
+
+ wait(MAX_IDLE_TIME, 5000, () -> !getCache().containsKey(KEY_ONE) && !getCache().containsKey(KEY_TWO));
+ }
+
+ @Test
+ default void publishMapNormalAsync() throws Exception {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.MAP, CollectionHelper.mapOf(KEY_ONE, VALUE_ONE, KEY_TWO, VALUE_TWO))
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTALLASYNC)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ assertEquals(VALUE_TWO, getCache().get(KEY_TWO));
+ }
+
+ @Test
+ default void publishMapWithLifespanAsync() throws Exception {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.MAP, CollectionHelper.mapOf(KEY_ONE, VALUE_ONE, KEY_TWO, VALUE_TWO))
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTALLASYNC)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_TIME)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ assertEquals(VALUE_TWO, getCache().get(KEY_TWO));
+
+ wait(LIFESPAN_TIME, 5000, () -> !getCache().containsKey(KEY_ONE) && !getCache().containsKey(KEY_TWO));
+ }
+
+ @Test
+ default void publishMapWithLifespanAndMaxIdleTimeAsync() throws Exception {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.MAP, CollectionHelper.mapOf(KEY_ONE, VALUE_ONE, KEY_TWO, VALUE_TWO))
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTALLASYNC)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_FOR_MAX_IDLE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME, MAX_IDLE_TIME)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ assertEquals(VALUE_TWO, getCache().get(KEY_TWO));
+
+ wait(MAX_IDLE_TIME, 5000, () -> !getCache().containsKey(KEY_ONE) && !getCache().containsKey(KEY_TWO));
+ }
+
+ @Test
+ default void putIfAbsentAlreadyExists() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTIFABSENT)
+ .send();
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ assertEquals(1, getCache().size());
+ }
+
+ @Test
+ default void putIfAbsentNotExists() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_TWO)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTIFABSENT)
+ .send();
+
+ assertEquals(VALUE_TWO, getCache().get(KEY_TWO));
+ assertEquals(2, getCache().size());
+ }
+
+ @Test
+ default void putIfAbsentKeyAndValueAsync() throws Exception {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTIFABSENTASYNC)
+ .request(CompletableFuture.class)
+ .get(1, TimeUnit.SECONDS);
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ }
+
+ @Test
+ default void putIfAbsentKeyAndValueAsyncWithLifespan() throws Exception {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTIFABSENTASYNC)
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_TIME)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+
+ wait(LIFESPAN_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void putIfAbsentKeyAndValueAsyncWithLifespanAndMaxIdle() throws Exception {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUTIFABSENTASYNC)
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_FOR_MAX_IDLE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME, MAX_IDLE_TIME)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ Thread.sleep(MAX_IDLE_TIME * 2);
+ assertFalse(getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void notContainsKeyTest() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ boolean result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_TWO)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.CONTAINSKEY)
+ .request(Boolean.class);
+
+ assertFalse(result);
+ }
+
+ @Test
+ default void containsKeyTest() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ boolean result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.CONTAINSKEY)
+ .request(Boolean.class);
+
+ assertTrue(result);
+ }
+
+ @Test
+ default void notContainsValueTest() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ boolean result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.CONTAINSVALUE)
+ .request(Boolean.class);
+
+ assertFalse(result);
+ }
+
+ @Test
+ default void containsValueTest() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ boolean result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.CONTAINSVALUE)
+ .request(Boolean.class);
+
+ assertTrue(result);
+ }
+
+ @Test
+ default void publishKeyAndValueWithLifespan() throws Exception {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_TIME)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUT)
+ .send();
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE).toString());
+ wait(LIFESPAN_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void getOrDefault() throws Exception {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUT)
+ .send();
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+
+ String result1 = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.DEFAULT_VALUE, "defaultTest")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.GETORDEFAULT)
+ .request(String.class);
+
+ assertEquals(VALUE_ONE, result1);
+
+ String result2 = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_TWO)
+ .withHeader(InfinispanConstants.DEFAULT_VALUE, "defaultTest")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.GETORDEFAULT)
+ .request(String.class);
+
+ assertEquals("defaultTest", result2);
+ }
+
+ @Test
+ default void putOperationReturnsThePreviousValue() throws Exception {
+ getCache().put(KEY_ONE, "existing value");
+
+ String result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUT)
+ .request(String.class);
+
+ assertEquals("existing value", result);
+ }
+
+ @Test
+ default void computeOperation() {
+ getCache().put(KEY_ONE, "existing value");
+
+ String result = fluentTemplate()
+ .to("direct:compute")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.COMPUTE)
+ .request(String.class);
+
+ assertEquals("existing valuereplay", result);
+ }
+
+ @Test
+ default void computeAsyncOperation() throws Exception {
+ getCache().put(KEY_ONE, "existing value");
+
+ CompletableFuture<?> result = fluentTemplate()
+ .to("direct:compute")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.COMPUTEASYNC)
+ .request(CompletableFuture.class);
+
+ assertEquals("existing valuereplay", result.get());
+ }
+
+ @Test
+ default void retrievesAValueByKey() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ String result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.GET)
+ .request(String.class);
+
+ assertEquals(VALUE_ONE, result);
+ }
+
+ @Test
+ default void replaceAValueByKey() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ String result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACE)
+ .request(String.class);
+
+ assertEquals(VALUE_ONE, result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+ }
+
+ @Test
+ default void replaceAValueByKeyWithLifespan() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ Object result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_TIME)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACE)
+ .request(String.class);
+
+ assertEquals(VALUE_ONE, result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+
+ wait(LIFESPAN_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void replaceAValueByKeyWithLifespanAndMaxIdleTime() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ Object result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_FOR_MAX_IDLE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME, MAX_IDLE_TIME)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACE)
+ .request(String.class);
+
+ assertEquals(VALUE_ONE, result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+
+ wait(MAX_IDLE_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void replaceAValueByKeyWithOldValue() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ Boolean result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.OLD_VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACE)
+ .request(Boolean.class);
+
+ assertTrue(result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+ }
+
+ @Test
+ default void replaceAValueByKeyWithLifespanWithOldValue() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ Object result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.OLD_VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_TIME)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACE)
+ .request(Boolean.class);
+
+ assertEquals(Boolean.TRUE, result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+
+ wait(LIFESPAN_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void replaceAValueByKeyWithLifespanAndMaxIdleTimeWithOldValue() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ Object result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.OLD_VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_FOR_MAX_IDLE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME, MAX_IDLE_TIME)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACE)
+ .request(Boolean.class);
+
+ assertEquals(Boolean.TRUE, result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+
+ wait(MAX_IDLE_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void replaceAValueByKeyAsync() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ String result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACEASYNC)
+ .request(String.class);
+
+ assertEquals(VALUE_ONE, result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+ }
+
+ @Test
+ default void replaceAValueByKeyWithLifespanAsync() throws Exception {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ Object result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_TIME)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACEASYNC)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(VALUE_ONE, result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+
+ wait(LIFESPAN_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void replaceAValueByKeyWithLifespanAndMaxIdleTimeAsync() throws Exception {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ Object result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_FOR_MAX_IDLE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME, MAX_IDLE_TIME)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACEASYNC)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(VALUE_ONE, result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+
+ wait(MAX_IDLE_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void replaceAValueByKeyAsyncWithOldValue() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ Boolean result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.OLD_VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACEASYNC)
+ .request(Boolean.class);
+
+ assertTrue(result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+ }
+
+ @Test
+ default void replaceAValueByKeyWithLifespanAsyncWithOldValue() throws Exception {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ Object result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.OLD_VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_TIME)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACEASYNC)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(Boolean.TRUE, result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+
+ wait(MAX_IDLE_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void replaceAValueByKeyWithLifespanAndMaxIdleTimeAsyncWithOldValue() throws Exception {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ Object result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_TWO)
+ .withHeader(InfinispanConstants.OLD_VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME, LIFESPAN_FOR_MAX_IDLE)
+ .withHeader(InfinispanConstants.LIFESPAN_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME, MAX_IDLE_TIME)
+ .withHeader(InfinispanConstants.MAX_IDLE_TIME_UNIT, TimeUnit.MILLISECONDS)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REPLACEASYNC)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertEquals(Boolean.TRUE, result);
+ assertEquals(VALUE_TWO, getCache().get(KEY_ONE));
+
+ wait(MAX_IDLE_TIME, 5000, () -> !getCache().containsKey(KEY_ONE));
+ }
+
+ @Test
+ default void deletesExistingValueByKey() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ String result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REMOVE)
+ .request(String.class);
+
+ assertEquals(VALUE_ONE, result);
+ assertNull(getCache().get(KEY_ONE));
+ }
+
+ @Test
+ default void deletesExistingValueByKeyAsync() throws Exception {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REMOVEASYNC)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertNull(getCache().get(KEY_ONE));
+ }
+
+ @Test
+ default void deletesExistingValueByKeyWithValue() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ Boolean result = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REMOVE)
+ .request(Boolean.class);
+
+ assertTrue(result);
+ assertNull(getCache().get(KEY_ONE));
+ }
+
+ @Test
+ default void deletesExistingValueByKeyAsyncWithValue() throws Exception {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REMOVEASYNC)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertNull(getCache().get(KEY_ONE));
+ }
+
+ @Test
+ default void clearsAllValues() {
+ getCache().put(KEY_ONE, VALUE_ONE);
+
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.CLEAR)
+ .send();
+
+ assertTrue(getCache().isEmpty());
+ }
+
+ @Test
+ default void testUriCommandOption() {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, COMMAND_KEY)
+ .withHeader(InfinispanConstants.VALUE, COMMAND_VALUE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUT)
+ .request(String.class);
+
+ assertEquals(COMMAND_VALUE, getCache().get(COMMAND_KEY));
+
+ String result2 = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, COMMAND_KEY)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.GET)
+ .request(String.class);
+
+ assertEquals(COMMAND_VALUE, result2);
+
+ String result3 = fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, COMMAND_KEY)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.REMOVE)
+ .request(String.class);
+
+ assertEquals(COMMAND_VALUE, result3);
+
+ assertNull(getCache().get(COMMAND_KEY));
+ assertTrue(getCache().isEmpty());
+
+ getCache().put(COMMAND_KEY, COMMAND_VALUE);
+ getCache().put("keyTest", "valueTest");
+
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.CLEAR)
+ .send();
+
+ assertTrue(getCache().isEmpty());
+ }
+
+ @Test
+ default void clearAsyncTest() throws Exception {
+ getCache().put(KEY_ONE, VALUE_ONE);
+ getCache().put(KEY_TWO, VALUE_TWO);
+
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.CLEARASYNC)
+ .request(CompletableFuture.class)
+ .get(5, TimeUnit.SECONDS);
+
+ assertTrue(getCache().isEmpty());
+ }
+
+ @Test
+ default void publishKeyAndValueByExplicitlySpecifyingTheKeyAndValueOptions() {
+ fluentTemplate()
+ .to("direct:explicitput")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUT)
+ .send();
+
+ assertEquals("3", getCache().get("a"));
+ }
+
+ @Test
+ default void publishKeyAndValueByExplicitlySpecifyingTheKeyAndValueOptionsHeaderHavePriorities() {
+ fluentTemplate()
+ .to("direct:explicitput")
+ .withHeader(InfinispanConstants.KEY, KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUT)
+ .send();
+
+ assertEquals(VALUE_ONE, getCache().get(KEY_ONE));
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteAggregationRepositoryIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteAggregationRepositoryIT.java
new file mode 100644
index 00000000000..ed2ebd038e6
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteAggregationRepositoryIT.java
@@ -0,0 +1,97 @@
+/*
+ * 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.infinispan.remote;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.annotation.DirtiesContext;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ InfinispanRemoteAggregationRepositoryIT.class
+ }
+)
+public class InfinispanRemoteAggregationRepositoryIT extends InfinispanRemoteTestSupport {
+ public static final int COMPLETION_SIZE = 4;
+ public static final String CORRELATOR_HEADER = "CORRELATOR_HEADER";
+
+ @Bean
+ public InfinispanRemoteAggregationRepository infinispanRemoteAggregationRepository() {
+ InfinispanRemoteConfiguration configuration = new InfinispanRemoteConfiguration();
+ configuration.setCacheContainerConfiguration(getConfiguration().build());
+
+ InfinispanRemoteAggregationRepository repo = new InfinispanRemoteAggregationRepository(getCacheName());
+ repo.setConfiguration(configuration);
+
+ return repo;
+ }
+
+ @Autowired
+ private ApplicationContext applicationContext;
+
+ @Test
+ public void checkAggregationFromOneRoute() throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .aggregate(header(CORRELATOR_HEADER))
+ .aggregationRepository(applicationContext.getBean(InfinispanRemoteAggregationRepository.class))
+ .aggregationStrategy((oldExchange, newExchange) -> {
+ if (oldExchange == null) {
+ return newExchange;
+ } else {
+ Integer n = newExchange.getIn().getBody(Integer.class);
+ Integer o = oldExchange.getIn().getBody(Integer.class);
+ Integer v = (o == null ? 0 : o) + (n == null ? 0 : n);
+ oldExchange.getIn().setBody(v, Integer.class);
+ return oldExchange;
+ }
+ })
+ .completionSize(COMPLETION_SIZE)
+ .to("mock:result");
+ }
+ });
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(2);
+ mock.expectedBodiesReceived(1 + 3 + 4 + 5, 6 + 7 + 20 + 21);
+
+ template.sendBodyAndHeader("direct:start", 1, CORRELATOR_HEADER, CORRELATOR_HEADER);
+ template.sendBodyAndHeader("direct:start", 3, CORRELATOR_HEADER, CORRELATOR_HEADER);
+ template.sendBodyAndHeader("direct:start", 4, CORRELATOR_HEADER, CORRELATOR_HEADER);
+ template.sendBodyAndHeader("direct:start", 5, CORRELATOR_HEADER, CORRELATOR_HEADER);
+ template.sendBodyAndHeader("direct:start", 6, CORRELATOR_HEADER, CORRELATOR_HEADER);
+ template.sendBodyAndHeader("direct:start", 7, CORRELATOR_HEADER, CORRELATOR_HEADER);
+ template.sendBodyAndHeader("direct:start", 20, CORRELATOR_HEADER, CORRELATOR_HEADER);
+ template.sendBodyAndHeader("direct:start", 21, CORRELATOR_HEADER, CORRELATOR_HEADER);
+
+ mock.assertIsSatisfied();
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteAggregationRepositoryOperationsIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteAggregationRepositoryOperationsIT.java
new file mode 100644
index 00000000000..a9b26fb716a
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteAggregationRepositoryOperationsIT.java
@@ -0,0 +1,246 @@
+/*
+ * 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.infinispan.remote;
+
+import java.util.Set;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.DefaultExchangeHolder;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.annotation.DirtiesContext;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ InfinispanRemoteAggregationRepositoryOperationsIT.class
+ }
+)
+public class InfinispanRemoteAggregationRepositoryOperationsIT extends InfinispanRemoteTestSupport {
+ private static InfinispanRemoteAggregationRepository aggregationRepository;
+
+ @BeforeAll
+ public static void setupResources() throws Exception {
+ InfinispanRemoteConfiguration configuration = new InfinispanRemoteConfiguration();
+ configuration.setCacheContainerConfiguration(getConfiguration().build());
+
+ aggregationRepository = new InfinispanRemoteAggregationRepository(InfinispanRemoteTestSupport.TEST_CACHE);
+ aggregationRepository.setConfiguration(configuration);
+ aggregationRepository.start();
+ }
+
+ @AfterAll
+ public static void cleanupResources() throws Exception {
+ if (aggregationRepository != null) {
+ aggregationRepository.stop();
+ }
+ }
+
+ private boolean exists(String key) {
+ DefaultExchangeHolder holder = aggregationRepository.getCache().get(key);
+ if (holder == null) {
+ return false;
+ }
+ return true;
+ }
+
+ @Test
+ public void testAdd() {
+ // cleanup
+ aggregationRepository.getCache().clear();
+ // Given
+ String key = "Add";
+ assertFalse(exists(key));
+ Exchange exchange = new DefaultExchange(context());
+ // When
+ aggregationRepository.add(context(), key, exchange);
+ // Then
+ assertTrue(exists(key));
+ }
+
+ @Test
+ public void testGetExists() {
+ // cleanup
+ aggregationRepository.getCache().clear();
+ // Given
+ String key = "Get_Exists";
+ Exchange exchange = new DefaultExchange(context());
+ aggregationRepository.add(context(), key, exchange);
+ assertTrue(exists(key));
+
+ // When
+ Exchange exchange2 = aggregationRepository.get(context(), key);
+ // Then
+ assertNotNull(exchange2);
+ assertEquals(exchange.getExchangeId(), exchange2.getExchangeId());
+ }
+
+ @Test
+ public void testGetNotExists() {
+ // cleanup
+ aggregationRepository.getCache().clear();
+ // Given
+ String key = "Get_NotExists";
+ assertFalse(exists(key));
+ // When
+ Exchange exchange2 = aggregationRepository.get(context(), key);
+ // Then
+ assertNull(exchange2);
+ }
+
+ @Test
+ public void testRemoveExists() {
+ // cleanup
+ aggregationRepository.getCache().clear();
+ // Given
+ String key = "Remove_Exists";
+ Exchange exchange = new DefaultExchange(context());
+ aggregationRepository.add(context(), key, exchange);
+ assertTrue(exists(key));
+ // When
+ aggregationRepository.remove(context(), key, exchange);
+ // Then
+ assertFalse(exists(key));
+ }
+
+ @Test
+ public void testRemoveNotExists() {
+ // cleanup
+ aggregationRepository.getCache().clear();
+ // Given
+ String key = "RemoveNotExists";
+ Exchange exchange = new DefaultExchange(context());
+ assertFalse(exists(key));
+ // When
+ aggregationRepository.remove(context(), key, exchange);
+ // Then
+ assertFalse(exists(key));
+ }
+
+ @Test
+ public void testGetKeys() {
+ // cleanup
+ aggregationRepository.getCache().clear();
+ // Given
+ String[] keys = { "GetKeys1", "GetKeys2" };
+ addExchanges(keys);
+ // When
+ Set<String> keySet = aggregationRepository.getKeys();
+ // Then
+ for (String key : keys) {
+ assertTrue(keySet.contains(key));
+ }
+ }
+
+ @Test
+ public void testConfirmExist() {
+ // cleanup
+ aggregationRepository.getCache().clear();
+ // Given
+ for (int i = 1; i < 4; i++) {
+ String key = "Confirm_" + i;
+ Exchange exchange = new DefaultExchange(context());
+ exchange.setExchangeId("Exchange_" + i);
+ aggregationRepository.add(context(), key, exchange);
+ assertTrue(exists(key));
+ }
+ // When
+ aggregationRepository.confirm(context(), "Confirm_2");
+ // Then
+ assertTrue(exists("Confirm_1"));
+ assertFalse(exists("Confirm_2"));
+ assertTrue(exists("Confirm_3"));
+ }
+
+ @Test
+ public void testConfirmNotExist() {
+ // cleanup
+ aggregationRepository.getCache().clear();
+ // Given
+ String[] keys = new String[3];
+ for (int i = 1; i < 4; i++) {
+ keys[i - 1] = "Confirm" + i;
+ }
+ addExchanges(keys);
+ for (String key : keys) {
+ assertTrue(exists(key));
+ }
+ // When
+ aggregationRepository.confirm(context(), "Exchange-Confirm5");
+ // Then
+ for (String key : keys) {
+ assertTrue(exists(key));
+ }
+ }
+
+ private void addExchanges(String... keys) {
+ // cleanup
+ aggregationRepository.getCache().clear();
+ for (String key : keys) {
+ Exchange exchange = new DefaultExchange(context());
+ exchange.setExchangeId("Exchange-" + key);
+ aggregationRepository.add(context(), key, exchange);
+ }
+ }
+
+ @Test
+ public void testScan() {
+ // cleanup
+ aggregationRepository.getCache().clear();
+ // Given
+ String[] keys = { "Scan1", "Scan2" };
+ addExchanges(keys);
+ // When
+ Set<String> exchangeIdSet = aggregationRepository.scan(context());
+ // Then
+ for (String key : keys) {
+ assertTrue(exchangeIdSet.contains(key));
+ }
+ }
+
+ @Test
+ public void testRecover() {
+ // cleanup
+ aggregationRepository.getCache().clear();
+ // Given
+ String[] keys = { "Recover1", "Recover2" };
+ addExchanges(keys);
+ // When
+ Exchange exchange2 = aggregationRepository.recover(context(), "Recover2");
+ Exchange exchange3 = aggregationRepository.recover(context(), "Recover3");
+ // Then
+ assertNotNull(exchange2);
+ assertNull(exchange3);
+ }
+
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteConfigurationIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteConfigurationIT.java
new file mode 100644
index 00000000000..f65c92fdc12
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteConfigurationIT.java
@@ -0,0 +1,120 @@
+/*
+ * 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.infinispan.remote;
+
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.test.infra.infinispan.services.InfinispanService;
+import org.apache.camel.test.infra.infinispan.services.InfinispanServiceFactory;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.infinispan.client.hotrod.RemoteCache;
+import org.infinispan.commons.api.BasicCache;
+import org.infinispan.configuration.cache.CacheMode;
+import org.jgroups.util.UUID;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.annotation.DirtiesContext;
+import org.testcontainers.shaded.org.apache.commons.lang3.SystemUtils;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ InfinispanRemoteConfigurationIT.class
+ }
+)
+public class InfinispanRemoteConfigurationIT {
+ @RegisterExtension
+ static InfinispanService service = InfinispanServiceFactory.createService();
+
+ @Test
+ public void remoteCacheWithoutProperties() throws Exception {
+ InfinispanRemoteConfiguration configuration = new InfinispanRemoteConfiguration();
+ configuration.setHosts(service.host() + ":" + service.port());
+ configuration.setSecure(true);
+ configuration.setUsername(service.username());
+ configuration.setPassword(service.password());
+ configuration.setSecurityServerName("infinispan");
+ configuration.setSaslMechanism("DIGEST-MD5");
+ configuration.setSecurityRealm("default");
+ if (SystemUtils.IS_OS_MAC) {
+ configuration.addConfigurationProperty(
+ "infinispan.client.hotrod.client_intelligence", "BASIC");
+ }
+
+ try (InfinispanRemoteManager manager = new InfinispanRemoteManager(configuration)) {
+ manager.start();
+ manager.getCacheContainer().administration()
+ .getOrCreateCache(
+ "misc_cache",
+ new org.infinispan.configuration.cache.ConfigurationBuilder()
+ .clustering()
+ .cacheMode(CacheMode.DIST_SYNC).build());
+
+ BasicCache<Object, Object> cache = manager.getCache("misc_cache");
+ assertNotNull(cache);
+ assertTrue(cache instanceof RemoteCache);
+
+ String key = UUID.randomUUID().toString();
+ assertNull(cache.put(key, "val1"));
+ assertNull(cache.put(key, "val2"));
+ }
+ }
+
+ @Test
+ public void remoteCacheWithProperties() throws Exception {
+ InfinispanRemoteConfiguration configuration = new InfinispanRemoteConfiguration();
+ configuration.setHosts(service.host() + ":" + service.port());
+ configuration.setSecure(true);
+ configuration.setUsername(service.username());
+ configuration.setPassword(service.password());
+ configuration.setSecurityServerName("infinispan");
+ configuration.setSaslMechanism("DIGEST-MD5");
+ configuration.setSecurityRealm("default");
+ if (SystemUtils.IS_OS_MAC) {
+ configuration.setConfigurationUri("infinispan/client-mac.properties");
+ } else {
+ configuration.setConfigurationUri("infinispan/client.properties");
+ }
+
+ try (InfinispanRemoteManager manager = new InfinispanRemoteManager(configuration)) {
+ manager.start();
+ manager.getCacheContainer().administration()
+ .getOrCreateCache(
+ "misc_cache",
+ new org.infinispan.configuration.cache.ConfigurationBuilder()
+ .clustering()
+ .cacheMode(CacheMode.DIST_SYNC).build());
+
+ BasicCache<Object, Object> cache = manager.getCache("misc_cache");
+ assertNotNull(cache);
+ assertTrue(cache instanceof RemoteCache);
+
+ String key = UUID.randomUUID().toString();
+ assertNull(cache.put(key, "val1"));
+ assertNotNull(cache.put(key, "val2"));
+ }
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteConsumerIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteConsumerIT.java
new file mode 100644
index 00000000000..5a923245ab4
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteConsumerIT.java
@@ -0,0 +1,90 @@
+/*
+ * 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.infinispan.remote;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.infinispan.InfinispanConstants;
+import org.apache.camel.component.infinispan.InfinispanConsumerTestSupport;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.infinispan.commons.api.BasicCache;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.annotation.DirtiesContext;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ InfinispanRemoteConsumerIT.class
+ }
+)
+public class InfinispanRemoteConsumerIT extends InfinispanRemoteTestSupport implements InfinispanConsumerTestSupport {
+ @Test
+ public void consumerReceivedEventNotifications() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+ mock.expectedHeaderReceived(InfinispanConstants.EVENT_TYPE, "CLIENT_CACHE_ENTRY_CREATED");
+
+ getCache().put(InfinispanConsumerTestSupport.KEY_ONE, InfinispanConsumerTestSupport.VALUE_ONE);
+ getCache().put(InfinispanConsumerTestSupport.KEY_ONE, InfinispanConsumerTestSupport.VALUE_TWO);
+
+ mock.assertIsSatisfied();
+ }
+
+ // *****************************
+ //
+ // *****************************
+
+ @BeforeEach
+ protected void beforeEach() {
+ // cleanup the default test cache before each run
+ getCache().clear();
+ }
+
+ @Override
+ public BasicCache<Object, Object> getCache() {
+ return super.getCache();
+ }
+
+ @Override
+ public BasicCache<Object, Object> getCache(String name) {
+ return super.getCache(name);
+ }
+
+ @Override
+ public MockEndpoint getMockEndpoint(String id) {
+ return super.getMockEndpoint(id);
+ }
+
+ @Bean
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ fromF("infinispan:%s?eventTypes=CLIENT_CACHE_ENTRY_CREATED", getCacheName())
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteCustomListenerIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteCustomListenerIT.java
new file mode 100644
index 00000000000..23c2cdf47fa
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteCustomListenerIT.java
@@ -0,0 +1,114 @@
+/*
+ * 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.infinispan.remote;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.infinispan.InfinispanConstants;
+import org.apache.camel.component.infinispan.InfinispanConsumerTestSupport;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.infinispan.client.hotrod.annotation.ClientCacheEntryCreated;
+import org.infinispan.client.hotrod.annotation.ClientListener;
+import org.infinispan.client.hotrod.event.ClientCacheEntryCreatedEvent;
+import org.infinispan.commons.api.BasicCache;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.annotation.DirtiesContext;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ InfinispanRemoteCustomListenerIT.class
+ }
+)
+public class InfinispanRemoteCustomListenerIT extends InfinispanRemoteTestSupport implements InfinispanConsumerTestSupport {
+ @Test
+ public void consumerReceivedEventNotifications() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+ mock.expectedHeaderReceived(InfinispanConstants.EVENT_TYPE, "CLIENT_CACHE_ENTRY_CREATED");
+
+ getCache().put(InfinispanConsumerTestSupport.KEY_ONE, InfinispanConsumerTestSupport.VALUE_ONE);
+ getCache().put(InfinispanConsumerTestSupport.KEY_ONE, InfinispanConsumerTestSupport.VALUE_TWO);
+
+ mock.assertIsSatisfied();
+ }
+
+ // *****************************
+ //
+ // *****************************
+
+ @BeforeEach
+ protected void beforeEach() {
+ // cleanup the default test cache before each run
+ getCache().clear();
+ }
+
+ @Override
+ public BasicCache<Object, Object> getCache() {
+ return super.getCache();
+ }
+
+ @Override
+ public BasicCache<Object, Object> getCache(String name) {
+ return super.getCache(name);
+ }
+
+ @Override
+ public MockEndpoint getMockEndpoint(String id) {
+ return super.getMockEndpoint(id);
+ }
+
+ @Bean
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ fromF("infinispan:%s?customListener=#myCustomListener", getCacheName())
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Bean
+ InfinispanRemoteCustomListener myCustomListener() {
+ return new InfinispanRemoteCustomListenerIT.MyRemoteCustomListener();
+ }
+
+ @ClientListener
+ public static class MyRemoteCustomListener extends InfinispanRemoteCustomListener {
+ @ClientCacheEntryCreated
+ public void entryCreated(ClientCacheEntryCreatedEvent<?> event) {
+ if (isAccepted(event.getType())) {
+ getEventProcessor().processEvent(
+ event.getType().toString(),
+ getCacheName(),
+ event.getKey(),
+ null,
+ null);
+ }
+ }
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteIdempotentRepositoryIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteIdempotentRepositoryIT.java
new file mode 100644
index 00000000000..42b3106b7eb
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteIdempotentRepositoryIT.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.component.infinispan.remote;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.infinispan.InfinispanIdempotentRepositoryTestSupport;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.IdempotentRepository;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.junit.jupiter.api.BeforeEach;
+
+import org.infinispan.commons.api.BasicCache;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.test.annotation.DirtiesContext;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ InfinispanRemoteIdempotentRepositoryIT.class
+ }
+)
+public class InfinispanRemoteIdempotentRepositoryIT extends InfinispanRemoteTestSupport
+ implements InfinispanIdempotentRepositoryTestSupport {
+
+ @Autowired
+ private ApplicationContext applicationContext;
+
+ @Bean
+ @Lazy
+ public IdempotentRepository repo() {
+ InfinispanRemoteIdempotentRepository repo = new InfinispanRemoteIdempotentRepository(getCacheName());
+ repo.setCacheContainer(cacheContainer);
+
+ return repo;
+ }
+
+ @BeforeEach
+ protected void beforeEach() {
+ // cleanup the default test cache before each run
+ getCache().clear();
+ }
+
+ @Override
+ public IdempotentRepository getIdempotentRepository() {
+ return applicationContext.getBean(IdempotentRepository.class);
+ }
+
+ @Override
+ public BasicCache<Object, Object> getCache() {
+ return super.getCache();
+ }
+
+ @Override
+ public MockEndpoint getMockEndpoint(String id) {
+ return super.getMockEndpoint(id);
+ }
+
+ @Bean
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .idempotentConsumer(
+ header("MessageID"),
+ getIdempotentRepository())
+ .skipDuplicate(true)
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteProducerIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteProducerIT.java
new file mode 100644
index 00000000000..792e04c41e5
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteProducerIT.java
@@ -0,0 +1,118 @@
+/*
+ * 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.infinispan.remote;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.infinispan.InfinispanConstants;
+import org.apache.camel.component.infinispan.InfinispanOperation;
+import org.apache.camel.component.infinispan.InfinispanProducerTestSupport;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.infinispan.client.hotrod.ServerStatistics;
+import org.infinispan.commons.api.BasicCache;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.annotation.DirtiesContext;
+
+import java.util.function.BiFunction;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ InfinispanRemoteProducerIT.class
+ }
+)
+public class InfinispanRemoteProducerIT extends InfinispanRemoteTestSupport implements InfinispanProducerTestSupport {
+
+ @Bean
+ public BiFunction<String, String, String> mappingFunction() {
+ return (k, v) -> v + "replay";
+ }
+
+ @Test
+ public void statsOperation() {
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, InfinispanProducerTestSupport.KEY_ONE)
+ .withHeader(InfinispanConstants.VALUE, InfinispanProducerTestSupport.VALUE_ONE)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUT)
+ .send();
+
+ assertEquals(InfinispanProducerTestSupport.VALUE_ONE, getCache().get(InfinispanProducerTestSupport.KEY_ONE));
+
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.KEY, InfinispanProducerTestSupport.KEY_TWO)
+ .withHeader(InfinispanConstants.VALUE, InfinispanProducerTestSupport.VALUE_TWO)
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.PUT)
+ .send();
+
+ assertEquals(InfinispanProducerTestSupport.VALUE_TWO, getCache().get(InfinispanProducerTestSupport.KEY_TWO));
+
+ assertEquals(
+ 2,
+ fluentTemplate()
+ .to("direct:start")
+ .withHeader(InfinispanConstants.OPERATION, InfinispanOperation.STATS)
+ .request(ServerStatistics.class)
+ .getIntStatistic(ServerStatistics.CURRENT_NR_OF_ENTRIES));
+ }
+
+ // *****************************
+ //
+ // *****************************
+
+ @BeforeEach
+ protected void beforeEach() {
+ // cleanup the default test cache before each run
+ getCache().clear();
+ }
+
+ @Override
+ public BasicCache<Object, Object> getCache() {
+ return super.getCache();
+ }
+
+ @Override
+ public BasicCache<Object, Object> getCache(String name) {
+ return InfinispanRemoteTestSupport.getCacheByName(name);
+ }
+
+ @Bean
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .toF("infinispan:%s", getCacheName());
+ from("direct:compute")
+ .toF("infinispan:%s?remappingFunction=#mappingFunction", getCacheName());
+ from("direct:explicitput")
+ .toF("infinispan:%s?operation=PUT&key=a&value=3", getCacheName());
+ }
+ };
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteQueryConsumerIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteQueryConsumerIT.java
new file mode 100644
index 00000000000..0e4c6b47c78
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteQueryConsumerIT.java
@@ -0,0 +1,157 @@
+/*
+ * 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.infinispan.remote;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.infinispan.InfinispanConstants;
+import org.apache.camel.component.infinispan.InfinispanQueryBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.infinispan.client.hotrod.marshall.MarshallerUtil;
+import org.infinispan.commons.api.BasicCache;
+import org.infinispan.commons.util.Util;
+import org.infinispan.protostream.FileDescriptorSource;
+import org.infinispan.protostream.SerializationContext;
+import org.infinispan.protostream.sampledomain.User;
+import org.infinispan.protostream.sampledomain.marshallers.GenderMarshaller;
+import org.infinispan.protostream.sampledomain.marshallers.UserMarshaller;
+import org.infinispan.query.remote.client.ProtobufMetadataManagerConstants;
+import org.infinispan.query.remote.client.impl.MarshallerRegistration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.annotation.DirtiesContext;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ InfinispanRemoteQueryConsumerIT.class
+ }
+)
+public class InfinispanRemoteQueryConsumerIT extends InfinispanRemoteQueryTestSupport {
+
+ @Bean
+ InfinispanQueryBuilder continuousQueryBuilderAll() {
+ return qf -> qf.from(User.class).having("name").like("%Q0%").build();
+ }
+
+ @Bean
+ InfinispanQueryBuilder continuousQueryBuilderNoMatch() {
+ return qf -> qf.from(User.class).having("name").like("%TEST%").build();
+ }
+
+ @Bean
+ InfinispanQueryBuilder continuousQueryBuilder() {
+ return qf -> qf.from(User.class).having("name").like("CQ%").build();
+ }
+
+ // *****************************
+ //
+ // *****************************
+
+ @Test
+ public void continuousQuery() throws Exception {
+ MockEndpoint continuousQueryBuilderNoMatch = getMockEndpoint("mock:continuousQueryNoMatch");
+ continuousQueryBuilderNoMatch.expectedMessageCount(0);
+
+ MockEndpoint continuousQueryBuilderAll = getMockEndpoint("mock:continuousQueryAll");
+ continuousQueryBuilderAll.expectedMessageCount(CQ_USERS.length * 2);
+
+ MockEndpoint continuousQuery = getMockEndpoint("mock:continuousQuery");
+ continuousQuery.expectedMessageCount(4);
+
+ for (int i = 0; i < 4; i++) {
+ continuousQuery.message(i).header(InfinispanConstants.KEY).isEqualTo(createKey(CQ_USERS[i % 2]));
+ continuousQuery.message(i).header(InfinispanConstants.CACHE_NAME).isEqualTo(getCache().getName());
+ if (i >= 2) {
+ continuousQuery.message(i).header(InfinispanConstants.EVENT_TYPE)
+ .isEqualTo(InfinispanConstants.CACHE_ENTRY_LEAVING);
+ continuousQuery.message(i).header(InfinispanConstants.EVENT_DATA).isNull();
+ } else {
+ continuousQuery.message(i).header(InfinispanConstants.EVENT_TYPE)
+ .isEqualTo(InfinispanConstants.CACHE_ENTRY_JOINING);
+ continuousQuery.message(i).header(InfinispanConstants.EVENT_DATA).isNotNull();
+ continuousQuery.message(i).header(InfinispanConstants.EVENT_DATA).isInstanceOf(User.class);
+ }
+ }
+
+ for (final User user : CQ_USERS) {
+ getCache().put(createKey(user), user);
+ }
+
+ assertEquals(CQ_USERS.length, getCache().size());
+
+ for (final User user : CQ_USERS) {
+ getCache().remove(createKey(user));
+ }
+
+ assertTrue(getCache().isEmpty());
+
+ continuousQuery.assertIsSatisfied();
+ continuousQueryBuilderNoMatch.assertIsSatisfied();
+ continuousQueryBuilderAll.assertIsSatisfied();
+ }
+
+ // *****************************
+ //
+ // *****************************
+
+ @BeforeAll
+ protected static void setupResources() throws Exception {
+ String proto = Util.read(InfinispanRemoteQueryConsumerIT.class.getResourceAsStream("/sample_bank_account/bank.proto"));
+
+ BasicCache<Object, Object> cache = getCacheByName(ProtobufMetadataManagerConstants.PROTOBUF_METADATA_CACHE_NAME);
+ cache.put("sample_bank_account/bank.proto", proto);
+
+ MarshallerRegistration.init(MarshallerUtil.getSerializationContext(cacheContainer));
+ SerializationContext serCtx = MarshallerUtil.getSerializationContext(cacheContainer);
+ serCtx.registerProtoFiles(FileDescriptorSource.fromResources("/sample_bank_account/bank.proto"));
+ serCtx.registerMarshaller(new UserMarshaller());
+ serCtx.registerMarshaller(new GenderMarshaller());
+ }
+
+ @BeforeEach
+ protected void beforeEach() {
+ // cleanup the default test cache before each run
+ getCache().clear();
+ }
+
+ @Bean
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ fromF("infinispan:%s?queryBuilder=#continuousQueryBuilder", getCacheName())
+ .to("mock:continuousQuery");
+ fromF("infinispan:%s?queryBuilder=#continuousQueryBuilderNoMatch", getCacheName())
+ .to("mock:continuousQueryNoMatch");
+ fromF("infinispan:%s?queryBuilder=#continuousQueryBuilderAll", getCacheName())
+ .to("mock:continuousQueryAll");
+ }
+ };
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteQueryProducerIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteQueryProducerIT.java
new file mode 100644
index 00000000000..190850f3f03
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteQueryProducerIT.java
@@ -0,0 +1,184 @@
+/*
+ * 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.infinispan.remote;
+
+import static org.apache.camel.component.infinispan.InfinispanConstants.OPERATION;
+import static org.apache.camel.component.infinispan.InfinispanConstants.QUERY_BUILDER;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.infinispan.InfinispanOperation;
+import org.apache.camel.component.infinispan.InfinispanQueryBuilder;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.support.DefaultMessage;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.infinispan.client.hotrod.configuration.ConfigurationBuilder;
+import org.infinispan.client.hotrod.marshall.MarshallerUtil;
+import org.infinispan.commons.api.BasicCache;
+import org.infinispan.commons.marshall.ProtoStreamMarshaller;
+import org.infinispan.commons.util.Util;
+import org.infinispan.protostream.FileDescriptorSource;
+import org.infinispan.protostream.SerializationContext;
+import org.infinispan.protostream.sampledomain.User;
+import org.infinispan.protostream.sampledomain.marshallers.GenderMarshaller;
+import org.infinispan.protostream.sampledomain.marshallers.UserMarshaller;
+import org.infinispan.query.remote.client.ProtobufMetadataManagerConstants;
+import org.infinispan.query.remote.client.impl.MarshallerRegistration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.annotation.DirtiesContext;
+
+import java.util.List;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ InfinispanRemoteQueryProducerIT.class
+ }
+)
+public class InfinispanRemoteQueryProducerIT extends InfinispanRemoteQueryTestSupport {
+
+ @Bean
+ public InfinispanQueryBuilder noResultQueryBuilder() {
+ return qf -> qf.from(User.class).having("name").like("%abc%").build();
+ }
+
+ @Bean
+ public InfinispanQueryBuilder withResultQueryBuilder() {
+ return qf -> qf.from(User.class).having("name").like("%A").build();
+ }
+
+ // *****************************
+ //
+ // *****************************
+
+ @Test
+ public void producerQueryOperationWithoutQueryBuilder() throws Exception {
+ Exchange request = template.request("direct:start",
+ exchange -> exchange.getIn().setHeader(OPERATION, InfinispanOperation.QUERY));
+ assertNull(request.getException());
+
+ List<?> queryResult = (List<?>) request.getIn().getBody();
+ assertNull(queryResult);
+ }
+
+ @Test
+ public void producerQueryWithoutResult() {
+ producerQueryWithoutResult("direct:start", noResultQueryBuilder());
+ }
+
+ @Test
+ public void producerQueryWithoutResultAndQueryBuilderFromConfig() {
+ producerQueryWithoutResult("direct:noQueryResults", null);
+ }
+
+ private void producerQueryWithoutResult(String endpoint, final InfinispanQueryBuilder builder) {
+ Exchange request = template.request(endpoint, createQueryProcessor(builder));
+
+ assertNull(request.getException());
+
+ List<User> queryResult = request.getIn().getBody(List.class);
+ assertNotNull(queryResult);
+ assertEquals(0, queryResult.size());
+ }
+
+ @Test
+ public void producerQueryWithResult() {
+ producerQueryWithResult("direct:start", withResultQueryBuilder());
+ }
+
+ @Test
+ public void producerQueryWithResultAndQueryBuilderFromConfig() {
+ producerQueryWithResult("direct:queryWithResults", null);
+ }
+
+ // *****************************
+ //
+ // *****************************
+
+ @BeforeAll
+ protected static void setupResources() throws Exception {
+ String proto = Util.read(InfinispanRemoteTestSupport.class.getResourceAsStream("/sample_bank_account/bank.proto"));
+
+ BasicCache<Object, Object> cache = getCacheByName(ProtobufMetadataManagerConstants.PROTOBUF_METADATA_CACHE_NAME);
+ cache.put("sample_bank_account/bank.proto", proto);
+
+ MarshallerRegistration.init(MarshallerUtil.getSerializationContext(cacheContainer));
+ SerializationContext serCtx = MarshallerUtil.getSerializationContext(cacheContainer);
+ serCtx.registerProtoFiles(FileDescriptorSource.fromResources("/sample_bank_account/bank.proto"));
+ serCtx.registerMarshaller(new UserMarshaller());
+ serCtx.registerMarshaller(new GenderMarshaller());
+ }
+
+ @BeforeEach
+ protected void beforeEach() {
+ // cleanup the default test cache before each run
+ getCache().clear();
+
+ for (final User user : USERS) {
+ getCache().put(createKey(user), user);
+ }
+ }
+
+ @Bean
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .toF("infinispan:%s", getCacheName());
+ from("direct:noQueryResults")
+ .toF("infinispan:%s?queryBuilder=#noResultQueryBuilder", getCacheName());
+ from("direct:queryWithResults")
+ .toF("infinispan:%s?queryBuilder=#withResultQueryBuilder", getCacheName());
+ }
+ };
+ }
+
+ private void producerQueryWithResult(String endpoint, final InfinispanQueryBuilder builder) {
+ Exchange request = template.request(endpoint, createQueryProcessor(builder));
+ assertNull(request.getException());
+
+ List<User> queryResult = request.getIn().getBody(List.class);
+ assertNotNull(queryResult);
+ assertEquals(2, queryResult.size());
+ assertTrue(hasUser(queryResult, "nameA", "surnameA"));
+ assertTrue(hasUser(queryResult, "nameA", "surnameB"));
+ }
+
+ private Processor createQueryProcessor(final InfinispanQueryBuilder builder) {
+ return exchange -> {
+ exchange.getIn().setHeader(OPERATION, InfinispanOperation.QUERY);
+ if (builder != null) {
+ exchange.getIn().setHeader(QUERY_BUILDER, builder);
+ }
+ };
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteQueryTestSupport.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteQueryTestSupport.java
new file mode 100644
index 00000000000..87b62f189c4
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteQueryTestSupport.java
@@ -0,0 +1,80 @@
+/*
+ * 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.infinispan.remote;
+
+import org.infinispan.protostream.sampledomain.User;
+
+import java.util.List;
+
+public class InfinispanRemoteQueryTestSupport extends InfinispanRemoteTestSupport {
+
+ public static final User[] USERS = new User[] {
+ createUser("nameA", "surnameA"),
+ createUser("nameA", "surnameB"),
+ createUser("nameB", "surnameB")};
+
+ public static final User[] CQ_USERS = new User[] {
+ createUser("CQ01", "surname01"),
+ createUser("CQ02", "surname01"),
+ createUser("NQ03", "surname03"),
+ createUser("NQ04", "surname04")
+ };
+
+ public static String createKey(User user) {
+ return String.format("%s+%s", user.getName(), user.getSurname());
+ }
+
+ public static User createUser(String name, String surname) {
+ User user = new User();
+ user.setName(name);
+ user.setSurname(surname);
+ return user;
+ }
+
+ public static boolean eq(String str1, String str2) {
+ if (str1 == null) {
+ return str2 == null;
+ } else {
+ return str1.equals(str2);
+ }
+ }
+
+ public static boolean eq(User user, String name, String surname) {
+ if (user == null) {
+ return false;
+ }
+ if (!eq(user.getName(), name)) {
+ return false;
+ }
+ if (!eq(user.getSurname(), surname)) {
+ return false;
+ }
+ return true;
+ }
+
+ public static boolean hasUser(List<User> users, String name, String surname) {
+ if (users == null) {
+ return false;
+ }
+ for (User user : users) {
+ if (eq(user, name, surname)) {
+ return true;
+ }
+ }
+ return false;
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteTestSupport.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteTestSupport.java
new file mode 100644
index 00000000000..4d0bbec4f29
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteTestSupport.java
@@ -0,0 +1,154 @@
+/*
+ * 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.infinispan.remote;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.FluentProducerTemplate;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.ComponentCustomizer;
+import org.apache.camel.test.infra.infinispan.services.InfinispanService;
+import org.apache.camel.test.infra.infinispan.services.InfinispanServiceFactory;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.TestMethodOrder;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import org.infinispan.client.hotrod.RemoteCacheManager;
+import org.infinispan.client.hotrod.configuration.ConfigurationBuilder;
+import org.infinispan.commons.api.BasicCache;
+import org.infinispan.commons.marshall.ProtoStreamMarshaller;
+import org.infinispan.configuration.cache.CacheMode;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Lazy;
+import org.testcontainers.shaded.org.apache.commons.lang3.SystemUtils;
+
+import java.util.Properties;
+
+@TestMethodOrder(MethodOrderer.MethodName.class)
+public class InfinispanRemoteTestSupport {
+ public static final String TEST_CACHE = "mycache";
+ @RegisterExtension
+ public static InfinispanService service = InfinispanServiceFactory.createService();
+ protected static RemoteCacheManager cacheContainer;
+
+ @Autowired
+ ProducerTemplate template;
+
+ @Autowired
+ CamelContext context;
+
+ protected BasicCache<Object, Object> getCache() {
+ return getCache(getCacheName());
+ }
+
+ protected String getCacheName() {
+ return TEST_CACHE;
+ }
+
+ protected MockEndpoint getMockEndpoint(String endpoint) {
+ return context.getEndpoint(endpoint, MockEndpoint.class);
+ }
+
+ public FluentProducerTemplate fluentTemplate() {
+ return context.createFluentProducerTemplate();
+ }
+
+ public CamelContext context() {
+ return context;
+ }
+
+ public ProducerTemplate template() {
+ return template;
+ }
+
+ @BeforeAll
+ protected static void setupResourcesAbstract() throws Exception {
+// LoggerFactory.getLogger(getClass()).info("setupResources");
+
+ cacheContainer = new RemoteCacheManager(getConfiguration().build());
+ cacheContainer.administration()
+ .getOrCreateCache(
+ TEST_CACHE,
+ new org.infinispan.configuration.cache.ConfigurationBuilder()
+ .clustering()
+ .cacheMode(CacheMode.DIST_SYNC).build());
+ }
+
+ @AfterAll
+ protected static void cleanupResourcesAbstract() throws Exception {
+// LoggerFactory.getLogger(getClass()).info("setupResources");
+
+ if (cacheContainer != null) {
+ cacheContainer.stop();
+ }
+ }
+
+ protected BasicCache<Object, Object> getCache(String name) {
+ return cacheContainer.getCache(name);
+ }
+
+ protected static BasicCache<Object, Object> getCacheByName(String name) {
+ return cacheContainer.getCache(name);
+ }
+
+ protected static ConfigurationBuilder getConfiguration() {
+ ConfigurationBuilder clientBuilder = new ConfigurationBuilder();
+
+ // for default tests, we force return value for all the
+ // operations
+ clientBuilder
+ .forceReturnValues(true);
+
+ // add server from the test infra service
+ clientBuilder
+ .addServer()
+ .host(service.host())
+ .port(service.port());
+
+ // add security info
+ clientBuilder
+ .security()
+ .authentication()
+ .username(service.username())
+ .password(service.password())
+ .serverName("infinispan")
+ .saslMechanism("DIGEST-MD5")
+ .realm("default");
+
+ if (SystemUtils.IS_OS_MAC) {
+ Properties properties = new Properties();
+ properties.put("infinispan.client.hotrod.client_intelligence", "BASIC");
+ clientBuilder.withProperties(properties);
+ }
+
+ clientBuilder.marshaller(new ProtoStreamMarshaller());
+
+ return clientBuilder;
+ }
+
+ @Lazy
+ @Bean
+ public ComponentCustomizer infinispanComponentCustomizer() throws Exception {
+ return ComponentCustomizer.forType(
+ InfinispanRemoteComponent.class,
+ component -> component.getConfiguration().setCacheContainer(cacheContainer));
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/cluster/InfinispanRemoteClusteredTestSupport.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/cluster/InfinispanRemoteClusteredTestSupport.java
new file mode 100644
index 00000000000..6b3da5d6c5e
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/cluster/InfinispanRemoteClusteredTestSupport.java
@@ -0,0 +1,74 @@
+/*
+ * 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.infinispan.remote.cluster;
+
+import java.util.Properties;
+
+import org.apache.camel.test.infra.infinispan.services.InfinispanService;
+
+import org.infinispan.client.hotrod.RemoteCacheManager;
+import org.infinispan.client.hotrod.configuration.Configuration;
+import org.infinispan.client.hotrod.configuration.ConfigurationBuilder;
+import org.infinispan.configuration.cache.CacheMode;
+import org.testcontainers.shaded.org.apache.commons.lang3.SystemUtils;
+
+public final class InfinispanRemoteClusteredTestSupport {
+ private InfinispanRemoteClusteredTestSupport() {
+ }
+
+ public static Configuration createConfiguration(InfinispanService service) {
+ if (SystemUtils.IS_OS_MAC) {
+ Properties properties = new Properties();
+ properties.put("infinispan.client.hotrod.client_intelligence", "BASIC");
+ return new ConfigurationBuilder()
+ .withProperties(properties)
+ .addServer()
+ .host(service.host())
+ .port(service.port())
+ .security()
+ .authentication()
+ .username(service.username())
+ .password(service.password())
+ .serverName("infinispan")
+ .saslMechanism("DIGEST-MD5")
+ .realm("default")
+ .build();
+ } else {
+ return new ConfigurationBuilder()
+ .addServer()
+ .host(service.host())
+ .port(service.port())
+ .security()
+ .authentication()
+ .username(service.username())
+ .password(service.password())
+ .serverName("infinispan")
+ .saslMechanism("DIGEST-MD5")
+ .realm("default")
+ .build();
+ }
+ }
+
+ public static void createCache(RemoteCacheManager cacheContainer, String cacheName) {
+ cacheContainer.administration()
+ .getOrCreateCache(
+ cacheName,
+ new org.infinispan.configuration.cache.ConfigurationBuilder()
+ .clustering()
+ .cacheMode(CacheMode.DIST_SYNC).build());
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/cluster/InfinispanRemoteClusteredViewIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/cluster/InfinispanRemoteClusteredViewIT.java
new file mode 100644
index 00000000000..a31b7b9d4a7
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/cluster/InfinispanRemoteClusteredViewIT.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.component.infinispan.remote.cluster;
+
+import static org.apache.camel.component.infinispan.remote.cluster.InfinispanRemoteClusteredTestSupport.createCache;
+import static org.apache.camel.component.infinispan.remote.cluster.InfinispanRemoteClusteredTestSupport.createConfiguration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.cluster.CamelClusterMember;
+import org.apache.camel.cluster.CamelClusterView;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.test.infra.infinispan.services.InfinispanService;
+import org.apache.camel.test.infra.infinispan.services.InfinispanServiceFactory;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import org.infinispan.client.hotrod.RemoteCacheManager;
+import org.infinispan.client.hotrod.configuration.Configuration;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.annotation.DirtiesContext;
+
+import java.util.concurrent.TimeUnit;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ InfinispanRemoteClusteredViewIT.class
+ }
+)
+public class InfinispanRemoteClusteredViewIT {
+ @RegisterExtension
+ public static InfinispanService service = InfinispanServiceFactory.createService();
+
+ @Autowired
+ private CamelContext context;
+
+ @Test
+ public void getLeaderTest() throws Exception {
+ final String viewName = "myView";
+
+ Configuration configuration = createConfiguration(service);
+
+ try (RemoteCacheManager cacheContainer = new RemoteCacheManager(configuration)) {
+ createCache(cacheContainer, viewName);
+
+ InfinispanRemoteClusterService clusterService = new InfinispanRemoteClusterService();
+ clusterService.setCacheContainer(cacheContainer);
+ clusterService.setId("node");
+
+ context.addService(clusterService);
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ fromF("master:%s:timer:infinispan?repeatCount=1", viewName)
+ .routeId("route1")
+ .stop();
+ }
+ });
+
+ CamelClusterView view = clusterService.getView(viewName);
+
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertThat(view.getLeader())
+ .get()
+ .satisfies(CamelClusterMember::isLeader)
+ .satisfies(CamelClusterMember::isLocal);
+ });
+ }
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/spring/SpringInfinispanRemoteIdempotentRepositoryCamelIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/spring/SpringInfinispanRemoteIdempotentRepositoryCamelIT.java
new file mode 100644
index 00000000000..6383b79d3a7
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/spring/SpringInfinispanRemoteIdempotentRepositoryCamelIT.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.component.infinispan.remote.spring;
+
+import org.apache.camel.component.infinispan.remote.InfinispanRemoteIdempotentRepository;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
+import org.springframework.test.annotation.DirtiesContext;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ SpringInfinispanRemoteIdempotentRepositoryCamelIT.class
+ },
+ properties = { "camel.springboot.routes-include-pattern=file:src/test/resources/org/apache/camel/component/infinispan/spring/SpringInfinispanRemoteIdempotentRepositoryCamelTest.xml" }
+)
+public class SpringInfinispanRemoteIdempotentRepositoryCamelIT
+ extends SpringInfinispanRemoteIdempotentRepositoryTestSupport {
+
+ @Bean
+ public org.apache.camel.spi.IdempotentRepository repo() {
+ InfinispanRemoteIdempotentRepository infinispanRemoteIdempotentRepository = new InfinispanRemoteIdempotentRepository("idempotent");
+ infinispanRemoteIdempotentRepository.setCacheContainer(manager);
+
+ return infinispanRemoteIdempotentRepository;
+ }
+
+ @Bean
+ public PropertySourcesPlaceholderConfigurer propertySourcesPlaceholderConfigurer() {
+ return new PropertySourcesPlaceholderConfigurer();
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/spring/SpringInfinispanRemoteIdempotentRepositorySpringIT.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/spring/SpringInfinispanRemoteIdempotentRepositorySpringIT.java
new file mode 100644
index 00000000000..07ca5fc49af
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/spring/SpringInfinispanRemoteIdempotentRepositorySpringIT.java
@@ -0,0 +1,68 @@
+/*
+ * 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.infinispan.remote.spring;
+
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.spring.processor.idempotent.SpringCacheIdempotentRepository;
+import org.apache.camel.test.spring.junit5.CamelSpringBootTest;
+
+import org.infinispan.spring.remote.provider.SpringRemoteCacheManagerFactoryBean;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.test.annotation.DirtiesContext;
+
+import java.util.Map;
+import java.util.Properties;
+
+@DirtiesContext
+@CamelSpringBootTest
+@SpringBootTest(
+ classes = {
+ CamelAutoConfiguration.class,
+ SpringInfinispanRemoteIdempotentRepositorySpringIT.class
+ },
+ properties = {"camel.springboot.routes-include-pattern=file:src/test/resources/org/apache/camel/component/infinispan/spring/SpringInfinispanRemoteIdempotentRepositorySpringTest.xml"}
+)
+public class SpringInfinispanRemoteIdempotentRepositorySpringIT
+ extends SpringInfinispanRemoteIdempotentRepositoryTestSupport {
+
+ @Bean
+ public SpringRemoteCacheManagerFactoryBean springRemoteCacheManagerFactoryBean() throws Exception {
+ Properties props = new Properties();
+ props.putAll(Map.of("infinispan.client.hotrod.server_list", service.getServiceAddress(),
+ "infinispan.client.hotrod.force_return_values", true,
+ "infinispan.client.hotrod.auth_server_name", "infinispan",
+ "infinispan.client.hotrod.auth_username", service.username(),
+ "infinispan.client.hotrod.auth_password", service.password(),
+ "infinispan.client.hotrod.auth_realm", "default",
+ "infinispan.client.hotrod.sasl_mechanism", "DIGEST-MD5"
+ ));
+ SpringRemoteCacheManagerFactoryBean springRemoteCacheManagerFactoryBean = new SpringRemoteCacheManagerFactoryBean();
+ springRemoteCacheManagerFactoryBean.setConfigurationProperties(props);
+ springRemoteCacheManagerFactoryBean.afterPropertiesSet();
+
+ return springRemoteCacheManagerFactoryBean;
+ }
+
+ @Bean
+ public SpringCacheIdempotentRepository repo() throws Exception {
+ SpringCacheIdempotentRepository springCacheIdempotentRepository =
+ new SpringCacheIdempotentRepository(springRemoteCacheManagerFactoryBean().getObject(), "idempotent");
+
+ return springCacheIdempotentRepository;
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/spring/SpringInfinispanRemoteIdempotentRepositoryTestSupport.java b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/spring/SpringInfinispanRemoteIdempotentRepositoryTestSupport.java
new file mode 100644
index 00000000000..6ffa486a36e
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/java/org/apache/camel/component/infinispan/remote/spring/SpringInfinispanRemoteIdempotentRepositoryTestSupport.java
@@ -0,0 +1,100 @@
+/*
+ * 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.infinispan.remote.spring;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.infinispan.remote.InfinispanRemoteIdempotentRepository;
+import org.apache.camel.component.infinispan.remote.InfinispanRemoteManager;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.IdempotentRepository;
+import org.apache.camel.test.infra.infinispan.services.InfinispanService;
+import org.apache.camel.test.infra.infinispan.services.InfinispanServiceFactory;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import org.infinispan.client.hotrod.RemoteCache;
+import org.infinispan.client.hotrod.RemoteCacheManager;
+import org.infinispan.client.hotrod.configuration.ConfigurationBuilder;
+import org.infinispan.client.hotrod.marshall.MarshallerUtil;
+import org.infinispan.query.remote.client.impl.MarshallerRegistration;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
+
+import java.util.UUID;
+
+public abstract class SpringInfinispanRemoteIdempotentRepositoryTestSupport {
+ @RegisterExtension
+ public static InfinispanService service = InfinispanServiceFactory.createService();
+
+ static RemoteCacheManager manager;
+
+ @Autowired
+ public CamelContext context;
+
+ @Autowired
+ public ProducerTemplate template;
+
+ @BeforeAll
+ public static void doPreSetup() throws Exception {
+ ConfigurationBuilder clientBuilder = new ConfigurationBuilder();
+
+ // for default tests, we force return value for all the
+ // operations
+ clientBuilder
+ .forceReturnValues(true);
+
+ // add server from the test infra service
+ clientBuilder
+ .addServer()
+ .host(service.host())
+ .port(service.port());
+
+ // add security info
+ clientBuilder
+ .security()
+ .authentication()
+ .username(service.username())
+ .password(service.password())
+ .serverName("infinispan")
+ .saslMechanism("DIGEST-MD5")
+ .realm("default");
+
+ manager = new RemoteCacheManager(clientBuilder.create());
+ MarshallerRegistration.init(MarshallerUtil.getSerializationContext(manager));
+ RemoteCache<Object, Object> cache = manager.administration().getOrCreateCache("idempotent", (String) null);
+ assertNotNull(cache);
+ }
+
+ @Test
+ public void testIdempotent() throws Exception {
+ MockEndpoint mock = context.getEndpoint("mock:result", MockEndpoint.class);
+ mock.expectedMessageCount(1);
+
+ String messageId = UUID.randomUUID().toString();
+ for (int i = 0; i < 5; i++) {
+ template.sendBodyAndHeader("direct:start", UUID.randomUUID().toString(), "MessageId", messageId);
+ }
+
+ mock.assertIsSatisfied();
+ }
+}
diff --git a/components-starter/camel-infinispan-starter/src/test/resources/infinispan/client-mac.properties b/components-starter/camel-infinispan-starter/src/test/resources/infinispan/client-mac.properties
new file mode 100644
index 00000000000..e4d6b3407bb
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/resources/infinispan/client-mac.properties
@@ -0,0 +1,24 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
+# Whether or not to implicitly Flag.FORCE_RETURN_VALUE for all calls.
+# The default value for this property is false.
+infinispan.client.hotrod.force_return_values = true
+
+
+# Enable development and test on Mac.
+infinispan.client.hotrod.client_intelligence=BASIC
diff --git a/components-starter/camel-infinispan-starter/src/test/resources/infinispan/client.properties b/components-starter/camel-infinispan-starter/src/test/resources/infinispan/client.properties
new file mode 100644
index 00000000000..ad7f1141e3a
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/resources/infinispan/client.properties
@@ -0,0 +1,20 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
+# Whether or not to implicitly Flag.FORCE_RETURN_VALUE for all calls.
+# The default value for this property is false.
+infinispan.client.hotrod.force_return_values = true
diff --git a/components-starter/camel-infinispan-starter/src/test/resources/log4j2.properties b/components-starter/camel-infinispan-starter/src/test/resources/log4j2.properties
new file mode 100644
index 00000000000..5a3ff87b317
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/resources/log4j2.properties
@@ -0,0 +1,48 @@
+## ---------------------------------------------------------------------------
+## 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-infinispan-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 = [%30.30t] %-30.30c{1} %-5p %m%n
+
+rootLogger.level = INFO
+rootLogger.appenderRef.file.ref = file
+#rootLogger.appenderRef.out.ref = out
+
+logger.infinispan-camel.name = org.apache.camel.component.infinispan
+logger.infinispan-camel.level = INFO
+logger.infinispan-camel-remote.name = org.apache.camel.component.infinispan.remote
+logger.infinispan-camel-remote.level = INFO
+logger.infinispan-camel-remote-cluster.name = org.apache.camel.component.infinispan.remote.cluster
+logger.infinispan-camel-remote-cluster.level = DEBUG
+logger.infinispan-camel-embedded.name = org.apache.camel.component.infinispan.embedded
+logger.infinispan-test-infra-container.name = container.infinispan
+logger.infinispan-test-infra-container.level = WARN
+logger.infinispan-camel-test.name = org.apache.camel.test.junit5
+logger.infinispan-camel-test.level = INFO
+
+logger.infinispan.name = org.infinispan
+logger.infinispan.level = INFO
+logger.jgroups.name = org.jgroups
+logger.jgroups.level = INFO
diff --git a/components-starter/camel-infinispan-starter/src/test/resources/org/apache/camel/component/infinispan/spring/SpringInfinispanRemoteIdempotentRepositoryCamelTest.xml b/components-starter/camel-infinispan-starter/src/test/resources/org/apache/camel/component/infinispan/spring/SpringInfinispanRemoteIdempotentRepositoryCamelTest.xml
new file mode 100644
index 00000000000..6411160b81c
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/resources/org/apache/camel/component/infinispan/spring/SpringInfinispanRemoteIdempotentRepositoryCamelTest.xml
@@ -0,0 +1,29 @@
+<?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 id="infinispan-remote">
+ <from uri="direct:start"/>
+ <idempotentConsumer idempotentRepository="repo" skipDuplicate="true">
+ <header>MessageId</header>
+ <to uri="log:org.apache.camel.component.infinispan?level=INFO&showAll=true&multiline=true"/>
+ <to uri="mock:result"/>
+ </idempotentConsumer>
+ </route>
+</routes>
\ No newline at end of file
diff --git a/components-starter/camel-infinispan-starter/src/test/resources/org/apache/camel/component/infinispan/spring/SpringInfinispanRemoteIdempotentRepositorySpringTest.xml b/components-starter/camel-infinispan-starter/src/test/resources/org/apache/camel/component/infinispan/spring/SpringInfinispanRemoteIdempotentRepositorySpringTest.xml
new file mode 100644
index 00000000000..63c8b1f7924
--- /dev/null
+++ b/components-starter/camel-infinispan-starter/src/test/resources/org/apache/camel/component/infinispan/spring/SpringInfinispanRemoteIdempotentRepositorySpringTest.xml
@@ -0,0 +1,29 @@
+<?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 id="infinispan-remote">
+ <from uri="direct:start"/>
+ <idempotentConsumer idempotentRepository="repo" skipDuplicate="true">
+ <header>MessageId</header>
+ <to uri="log:org.apache.camel.component.infinispan?level=INFO&showAll=true&multiline=true"/>
+ <to uri="mock:result"/>
+ </idempotentConsumer>
+ </route>
+</routes>