payloadMatches(String topic, Predicate... payloadMatchers) {
for (Predicate
payloadMatcher : payloadMatchers) {
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainerTestUtils.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainerTestUtils.java
new file mode 100644
index 00000000..7686ea8a
--- /dev/null
+++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainerTestUtils.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://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.springframework.cloud.stream.app.test.integration;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.util.SocketUtils;
+
+/**
+ * Support utility for stream application integration testing .
+ * @author David Turanski
+ */
+public abstract class StreamAppContainerTestUtils {
+
+ /**
+ * Default docker org.
+ */
+ public static String DOCKER_ORG = "springcloudstream";
+
+ public static final String prePackagedStreamAppImageName(String appName, String binderName, String version) {
+ return DOCKER_ORG + "/" + appName + "-" + binderName + ":" + version;
+ }
+
+ public static final String localHostAddress() {
+ try {
+ return InetAddress.getLocalHost().getHostAddress();
+ }
+ catch (UnknownHostException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+
+ public static final File resourceAsFile(String path) {
+ try {
+ return new ClassPathResource(path).getFile();
+ }
+ catch (IOException e) {
+ throw new IllegalStateException("Unable to access resource " + path);
+ }
+ }
+
+ public static final int findAvailablePort() {
+ return SocketUtils.findAvailableTcpPort(10000, 20000);
+ }
+}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/TestTopicSender.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/TestTopicSender.java
new file mode 100644
index 00000000..d86331fb
--- /dev/null
+++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/TestTopicSender.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://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.springframework.cloud.stream.app.test.integration;
+
+import org.springframework.messaging.Message;
+
+public interface TestTopicSender {
+
void send(String topic, P payload);
+
+ void send(String topic, Message message);
+}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaConfig.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaConfig.java
new file mode 100644
index 00000000..148de7b2
--- /dev/null
+++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaConfig.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://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.springframework.cloud.stream.app.test.integration.kafka;
+
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.utility.DockerImageName;
+
+import org.springframework.cloud.stream.app.test.integration.StreamAppContainer;
+import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils;
+
+public abstract class KafkaConfig {
+ final static String BINDER = "kafka";
+
+ final static Network network = Network.SHARED;
+
+ public static StreamAppContainer prepackagedContainerFor(String appName, String version) {
+ return new KafkaStreamAppContainer(
+ StreamAppContainerTestUtils.prePackagedStreamAppImageName(appName, BINDER, version),
+ kafka);
+ }
+
+ /**
+ * The KafkaContainer.
+ */
+ public final static KafkaContainer kafka = new KafkaContainer(
+ DockerImageName.parse("confluentinc/cp-kafka:5.5.1"))
+ .withExposedPorts(9092, 9093)
+ .withNetwork(network);
+
+ static {
+ kafka.start();
+ }
+
+}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainerTestConfiguration.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainerTestConfiguration.java
new file mode 100644
index 00000000..e0a6b2e3
--- /dev/null
+++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainerTestConfiguration.java
@@ -0,0 +1,209 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://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.springframework.cloud.stream.app.test.integration.kafka;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import java.util.function.Function;
+
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.KafkaAdminClient;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import org.springframework.cloud.stream.app.test.integration.AbstractTestTopicListener;
+import org.springframework.cloud.stream.app.test.integration.MessageMatcher;
+import org.springframework.cloud.stream.app.test.integration.OutputMatcher;
+import org.springframework.cloud.stream.app.test.integration.TestTopicListener;
+import org.springframework.cloud.stream.app.test.integration.TestTopicSender;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.annotation.KafkaHandler;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.Message;
+
+import static org.awaitility.Awaitility.await;
+import static org.springframework.cloud.stream.app.test.integration.AbstractTestTopicListener.STREAM_APPLICATIONS_TEST_TOPIC;
+import static org.springframework.cloud.stream.app.test.integration.kafka.KafkaConfig.kafka;
+
+@Configuration
+@EnableKafka
+public class KafkaStreamAppContainerTestConfiguration {
+
+ private static final String SUFFIX = UUID.randomUUID().toString().substring(0, 8);
+
+ private static final String STREAM_APPLICATION_TESTS_GROUP = "stream-application-tests_" + SUFFIX;
+
+ @Bean
+ public KafkaTemplate kafkaTemplate(ProducerFactory producerFactory) {
+ return new KafkaTemplate(producerFactory);
+ }
+
+ @Bean
+ public OutputMatcher outputMatcher(TestTopicListener testTopicListener) {
+ return new OutputMatcher(testTopicListener);
+ }
+
+ @Bean
+ public TestTopicSender testTopicSender(KafkaTemplate kafkaTemplate) {
+ return new KafkaTemplateTopicSender(kafkaTemplate);
+ }
+
+ @Bean
+ public ConsumerFactory consumerFactory() {
+ Map configs = new HashMap<>();
+ configs.put(ConsumerConfig.GROUP_ID_CONFIG, STREAM_APPLICATION_TESTS_GROUP);
+ configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(configs);
+ cf.setBootstrapServersSupplier(() -> kafka.getBootstrapServers());
+ return cf;
+ }
+
+ @Bean
+ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory(
+ ConsumerFactory consumerFactory) {
+
+ ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
+ factory.setConsumerFactory(consumerFactory);
+ return factory;
+ }
+
+ @Bean
+ public ProducerFactory producerFactory() {
+ Map configs = new HashMap<>();
+ configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(configs);
+ pf.setBootstrapServersSupplier(() -> kafka.getBootstrapServers());
+ return pf;
+ }
+
+ @Bean
+ public AdminClient admin() {
+ Map configs = new HashMap<>();
+ configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
+ return KafkaAdminClient.create(configs);
+ }
+
+ @Bean
+ public KafkaTestListener testListener(AdminClient admin, KafkaListenerEndpointRegistry endpointRegistry) {
+ return new KafkaTestListener(admin, endpointRegistry);
+ }
+
+ @KafkaListener(autoStartup = "true", topicPattern = STREAM_APPLICATIONS_TEST_TOPIC)
+ static class KafkaTestListener extends AbstractTestTopicListener {
+
+ private final AdminClient admin;
+
+ private final KafkaListenerEndpointRegistry endpointRegistry;
+
+ private final Object lock = new Object();
+
+ KafkaTestListener(AdminClient admin, KafkaListenerEndpointRegistry endpointRegistry) {
+ super();
+ this.admin = admin;
+ this.endpointRegistry = endpointRegistry;
+ this.admin.createTopics(
+ Collections.singletonList(
+ new NewTopic(STREAM_APPLICATIONS_TEST_TOPIC, Optional.empty(), Optional.empty())));
+ await().atMost(Duration.ofSeconds(30))
+ .until(() -> {
+ Set topics = admin.listTopics().names().get();
+ return topics.contains(STREAM_APPLICATIONS_TEST_TOPIC);
+ });
+ }
+
+ @Override
+ public boolean addMessageMatcher(String topic, MessageMatcher messageMatcher) {
+ boolean added = super.addMessageMatcher(topic, messageMatcher);
+ if (added) {
+ synchronized (lock) {
+ stop();
+ // rewind to consume messages that may have arrived before a verifier is registered.
+ admin.alterConsumerGroupOffsets(STREAM_APPLICATION_TESTS_GROUP,
+ Collections.singletonMap(new TopicPartition(topic, 0), new OffsetAndMetadata(0)));
+ start();
+ }
+ }
+ return added;
+ }
+
+ private void stop() {
+ this.endpointRegistry.getAllListenerContainers().forEach(container -> container.stop());
+ }
+
+ private void start() {
+ this.endpointRegistry.getAllListenerContainers().forEach(container -> container.start());
+ }
+
+ @Override
+ protected Function, String> topicForMessage() {
+ return message -> (String) message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC);
+ }
+
+ @KafkaHandler(isDefault = true)
+ public void listen(Message message) {
+ super.listen(message);
+ }
+ }
+
+ static class KafkaTemplateTopicSender implements TestTopicSender {
+
+ private final KafkaTemplate kafkaTemplate;
+
+ KafkaTemplateTopicSender(KafkaTemplate kafkaTemplate) {
+ this.kafkaTemplate = kafkaTemplate;
+ }
+
+ @Override
+ public void send(String topic, P payload) {
+ doSend(topic, payload);
+ }
+
+ @Override
+ public void send(String topic, Message message) {
+ doSend(topic, message);
+ }
+
+ private void doSend(String topic, Object payload) {
+ kafkaTemplate.send(topic, payload);
+ }
+ }
+
+}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppTest.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppTest.java
new file mode 100644
index 00000000..3c45d213
--- /dev/null
+++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppTest.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://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.springframework.cloud.stream.app.test.integration.kafka;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Testcontainers
+@ExtendWith(SpringExtension.class)
+@ContextConfiguration(classes = KafkaStreamAppContainerTestConfiguration.class)
+public @interface KafkaStreamAppTest {
+}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupport.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupport.java
deleted file mode 100644
index 89ef68af..00000000
--- a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupport.java
+++ /dev/null
@@ -1,205 +0,0 @@
-/*
- * Copyright 2020-2020 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * https://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.springframework.cloud.stream.app.test.integration.kafka;
-
-import java.time.Duration;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.UUID;
-import java.util.function.Function;
-
-import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.AdminClientConfig;
-import org.apache.kafka.clients.admin.KafkaAdminClient;
-import org.apache.kafka.clients.admin.NewTopic;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.OffsetAndMetadata;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.serialization.StringDeserializer;
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.testcontainers.containers.KafkaContainer;
-import org.testcontainers.containers.Network;
-import org.testcontainers.utility.DockerImageName;
-
-import org.springframework.cloud.stream.app.test.integration.AbstractTestTopicListener;
-import org.springframework.cloud.stream.app.test.integration.MessageMatcher;
-import org.springframework.cloud.stream.app.test.integration.StreamAppContainer;
-import org.springframework.cloud.stream.app.test.integration.StreamApplicationIntegrationTestSupport;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.kafka.annotation.EnableKafka;
-import org.springframework.kafka.annotation.KafkaHandler;
-import org.springframework.kafka.annotation.KafkaListener;
-import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
-import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
-import org.springframework.kafka.core.ConsumerFactory;
-import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
-import org.springframework.kafka.core.DefaultKafkaProducerFactory;
-import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.core.ProducerFactory;
-import org.springframework.kafka.support.KafkaHeaders;
-import org.springframework.messaging.Message;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit.jupiter.SpringExtension;
-
-import static org.awaitility.Awaitility.await;
-import static org.springframework.cloud.stream.app.test.integration.AbstractTestTopicListener.STREAM_APPLICATIONS_TEST_TOPIC;
-
-/**
- * Base class for stream application integration testing with Test Containers and Kafka
- * binder.
- */
-@ExtendWith(SpringExtension.class)
-@ContextConfiguration(classes = KafkaStreamApplicationIntegrationTestSupport.KafkaTestConfiguration.class)
-public abstract class KafkaStreamApplicationIntegrationTestSupport extends StreamApplicationIntegrationTestSupport {
-
- final static String BINDER = "kafka";
-
- final static Network network = Network.SHARED;
-
- protected final static KafkaContainer kafka = new KafkaContainer(
- DockerImageName.parse("confluentinc/cp-kafka:5.5.1"))
- .withExposedPorts(9092, 9093)
- .withNetwork(network);
-
- static {
- kafka.start();
- }
-
- protected static StreamAppContainer prepackagedKafkaContainerFor(String appName, String version) {
- return new KafkaStreamAppContainer(prePackagedStreamAppImageName(appName, BINDER, version),
- kafka);
- }
-
- @Configuration
- @EnableKafka
- static class KafkaTestConfiguration {
- private static final String SUFFIX = UUID.randomUUID().toString().substring(0, 8);
-
- private static final String STREAM_APPLICATION_TESTS_GROUP = "stream-application-tests_" + SUFFIX;
-
- @Bean
- KafkaTemplate kafkaTemplate(ProducerFactory producerFactory) {
- return new KafkaTemplate(producerFactory);
- }
-
- @Bean
- public ConsumerFactory consumerFactory() {
- Map configs = new HashMap<>();
- configs.put(ConsumerConfig.GROUP_ID_CONFIG, STREAM_APPLICATION_TESTS_GROUP);
- configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
- configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
- configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
- DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(configs);
- cf.setBootstrapServersSupplier(() -> kafka.getBootstrapServers());
- return cf;
- }
-
- @Bean
- public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory(
- ConsumerFactory consumerFactory) {
-
- ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
- factory.setConsumerFactory(consumerFactory);
- return factory;
- }
-
- @Bean
- public ProducerFactory producerFactory() {
- Map configs = new HashMap<>();
- configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
- configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
- DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(configs);
- pf.setBootstrapServersSupplier(() -> kafka.getBootstrapServers());
- return pf;
- }
-
- @Bean
- public AdminClient admin() {
- Map configs = new HashMap<>();
- configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
- return KafkaAdminClient.create(configs);
- }
-
- @Bean
- KafkaTestListener testListener(AdminClient admin, KafkaListenerEndpointRegistry endpointRegistry) {
- return new KafkaTestListener(admin, endpointRegistry);
- }
-
- @KafkaListener(autoStartup = "true", topicPattern = STREAM_APPLICATIONS_TEST_TOPIC)
- static class KafkaTestListener extends AbstractTestTopicListener {
-
- private final AdminClient admin;
-
- private final KafkaListenerEndpointRegistry endpointRegistry;
-
- private final Object lock = new Object();
-
- KafkaTestListener(AdminClient admin, KafkaListenerEndpointRegistry endpointRegistry) {
- super();
- this.admin = admin;
- this.endpointRegistry = endpointRegistry;
- this.admin.createTopics(
- Collections.singletonList(
- new NewTopic(STREAM_APPLICATIONS_TEST_TOPIC, Optional.empty(), Optional.empty())));
- await().atMost(Duration.ofSeconds(30))
- .until(() -> {
- Set topics = admin.listTopics().names().get();
- return topics.contains(STREAM_APPLICATIONS_TEST_TOPIC);
- });
- }
-
- @Override
- public boolean addMessageMatcher(String topic, MessageMatcher messageMatcher) {
- boolean added = super.addMessageMatcher(topic, messageMatcher);
- if (added) {
- synchronized (lock) {
- stop();
- // rewind to consume messages that may have arrived before a verifier is registered.
- admin.alterConsumerGroupOffsets(STREAM_APPLICATION_TESTS_GROUP,
- Collections.singletonMap(new TopicPartition(topic, 0), new OffsetAndMetadata(0)));
- start();
- }
- }
- return added;
- }
-
- private void stop() {
- this.endpointRegistry.getAllListenerContainers().forEach(container -> container.stop());
- }
-
- private void start() {
- this.endpointRegistry.getAllListenerContainers().forEach(container -> container.start());
- }
-
- @Override
- protected Function, String> topicForMessage() {
- return message -> (String) message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC);
- }
-
- @KafkaHandler(isDefault = true)
- public void listen(Message message) {
- super.listen(message);
- }
- }
- }
-}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQConfig.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQConfig.java
new file mode 100644
index 00000000..cc6aa388
--- /dev/null
+++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQConfig.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://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.springframework.cloud.stream.app.test.integration.rabbitmq;
+
+import org.testcontainers.containers.Network;
+import org.testcontainers.containers.RabbitMQContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import org.springframework.cloud.stream.app.test.integration.StreamAppContainer;
+import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils;
+
+public abstract class RabbitMQConfig {
+ /**
+ * The RabbitMQContainer.
+ */
+ public static RabbitMQContainer rabbitmq;
+
+ final static String BINDER = "rabbit";
+
+ final static Network network = Network.SHARED;
+
+ static {
+ rabbitmq = new RabbitMQContainer(DockerImageName.parse("rabbitmq:3"))
+ .withNetwork(network)
+ .withExposedPorts(5672, 15672);
+ rabbitmq.start();
+ }
+
+ public static StreamAppContainer prepackagedContainerFor(String appName, String version) {
+ return new RabbitMQStreamAppContainer(
+ StreamAppContainerTestUtils.prePackagedStreamAppImageName(appName, BINDER, version),
+ rabbitmq);
+ }
+
+}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainerTestConfiguration.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainerTestConfiguration.java
new file mode 100644
index 00000000..472dcff7
--- /dev/null
+++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainerTestConfiguration.java
@@ -0,0 +1,276 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://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.springframework.cloud.stream.app.test.integration.rabbitmq;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
+import java.util.function.Predicate;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.annotation.EnableRabbit;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.cloud.stream.app.test.integration.AbstractTestTopicListener;
+import org.springframework.cloud.stream.app.test.integration.MessageMatcher;
+import org.springframework.cloud.stream.app.test.integration.OutputMatcher;
+import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils;
+import org.springframework.cloud.stream.app.test.integration.TestTopicListener;
+import org.springframework.cloud.stream.app.test.integration.TestTopicSender;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.lang.NonNull;
+import org.springframework.messaging.Message;
+import org.springframework.retry.backoff.FixedBackOffPolicy;
+import org.springframework.retry.policy.SimpleRetryPolicy;
+import org.springframework.retry.support.RetryTemplate;
+import org.springframework.util.CollectionUtils;
+
+@Configuration
+@EnableRabbit
+public abstract class RabbitMQStreamAppContainerTestConfiguration {
+
+ private static final String STREAM_APPLICATION_TESTS_GROUP = "stream-application-tests";
+
+ @Bean
+ public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
+ RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
+ RetryTemplate retryTemplate = new RetryTemplate();
+ FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy();
+ fixedBackOffPolicy.setBackOffPeriod(2000);
+ SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
+ retryPolicy.setMaxAttempts(5);
+ retryTemplate.setBackOffPolicy(fixedBackOffPolicy);
+ retryTemplate.setRetryPolicy(retryPolicy);
+ rabbitTemplate.setRetryTemplate(retryTemplate);
+ // Need to be synchronous in case the app input exchange isn't ready to receive messages
+ rabbitTemplate.setChannelTransacted(true);
+ return rabbitTemplate;
+
+ }
+
+ @Bean
+ public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
+ return new RabbitAdmin(connectionFactory);
+ }
+
+ @Bean
+ public OutputMatcher outputMatcher(TestTopicListener testTopicListener) {
+ return new OutputMatcher(testTopicListener);
+ }
+
+ @Bean
+ public TestTopicSender testTopicSender(RabbitTemplate rabbitTemplate) {
+ return new RabbitTemplateTopicSender(rabbitTemplate);
+ }
+
+ @Bean
+ public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
+ ConnectionFactory connectionFactory) {
+ SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
+ factory.setConnectionFactory(connectionFactory);
+ factory.setMessageConverter(new MessageConverter() {
+ @Override
+ public org.springframework.amqp.core.Message toMessage(Object o, MessageProperties messageProperties)
+ throws MessageConversionException {
+ throw new UnsupportedOperationException("toMessage not implemented.");
+ }
+
+ @Override
+ public Object fromMessage(org.springframework.amqp.core.Message message)
+ throws MessageConversionException {
+ return new String(message.getBody());
+ }
+ });
+ return factory;
+ }
+
+ @Bean
+ public ConnectionFactory connectionFactory() {
+ return new CachingConnectionFactory(StreamAppContainerTestUtils.localHostAddress(),
+ RabbitMQConfig.rabbitmq.getMappedPort(5672));
+ }
+
+ @Bean
+ public RabbitMQTestListener rabbitMQTestListener(RabbitAdmin admin) {
+ return new RabbitMQTestListener(admin);
+ }
+
+ static class RabbitMQTestListener extends AbstractTestTopicListener {
+
+ public static final int CACHE_TTL_SEC = 120;
+
+ private final Cache>> cache = Caffeine.newBuilder()
+ .expireAfterWrite(CACHE_TTL_SEC, TimeUnit.SECONDS)
+ .build();
+
+ static final String STREAM_APPLICATIONS_TEST_QUEUE = "stream-applications-test-queue";
+
+ private final RabbitAdmin admin;
+
+ private final Queue queue;
+
+ private final TopicExchange exchange = new TopicExchange(STREAM_APPLICATIONS_TEST_TOPIC);
+
+ RabbitMQTestListener(RabbitAdmin admin) {
+ super();
+ this.admin = admin;
+ this.queue = new Queue(STREAM_APPLICATIONS_TEST_QUEUE);
+ admin.declareQueue(queue);
+ admin.declareExchange(exchange);
+ admin.declareBinding(
+ BindingBuilder.bind(queue).to(exchange).with("#"));
+ }
+
+ @Override
+ public AtomicBoolean allMatch(String topic) {
+ AtomicBoolean all = super.allMatch(topic);
+ if (!all.get()) {
+ if (messageMatchers.get(topic) == null || cache.getIfPresent(topic) == null) {
+ return all;
+ }
+ List matchers = messageMatchers.get(topic);
+ all.set(true);
+ cache.getIfPresent(topic).forEach(message -> matchers.stream().filter(mm -> !mm.isSatisfied())
+ .forEach(mm -> all.compareAndSet(true, mm.test(message))));
+ }
+ return all;
+ }
+
+ @Override
+ public AtomicBoolean matches(String topic, Predicate... predicates) {
+ AtomicBoolean matches = super.matches(topic, predicates);
+ if (matches.get() || CollectionUtils.isEmpty(cache.getIfPresent(topic))) {
+ return matches;
+ }
+
+ for (Predicate predicate : predicates) {
+ MessageMatcher matcher = messageMatcher(topic, predicate)
+ .orElse(MessageMatcher.payloadMatcher(o -> false));
+ if (messageMatcher(topic, predicate).isPresent()) {
+
+ Set> messages = cache.getIfPresent(topic);
+
+ matches.set(true);
+ messages.forEach(message -> {
+ if (matches.compareAndSet(true, matcher.test(message))) {
+ logger.debug("Matched cached message {} for topic {}", message, topic);
+ return;
+ }
+ });
+ }
+ }
+ return matches;
+ }
+
+ private void updateCache(String topic, Set> messages) {
+ if (CollectionUtils.isEmpty(messages)) {
+ cache.invalidate(topic);
+ }
+ else {
+ cache.put(topic, messages);
+ }
+ }
+
+ private void cacheMessage(String topic, Message message) {
+ if (cache.getIfPresent(topic) == null) {
+ cache.put(topic, new HashSet<>());
+ }
+ Set> messages = cache.getIfPresent(topic);
+ if (messages.add(message)) {
+ logger.debug("Caching message: {} for topic {}", message, topic);
+ }
+
+ }
+
+ @Override
+ protected Function, String> topicForMessage() {
+ return message -> (String) message.getHeaders().get(AmqpHeaders.RECEIVED_EXCHANGE);
+ }
+
+ @RabbitListener(autoStartup = "true", group = STREAM_APPLICATION_TESTS_GROUP, queues = {
+ STREAM_APPLICATIONS_TEST_QUEUE
+ })
+ @Override
+ public void listen(Message message) {
+ String topic = topicForMessage().apply(message);
+ logger.debug("Received message: {} on topic {}", message, topic);
+ if (!messageMatchers.containsKey(topic)) {
+ cacheMessage(topic, message);
+ return;
+ }
+
+ logger.debug("Verifying message: {} on topic {}", message, topic);
+ AtomicBoolean any = new AtomicBoolean(false);
+ messageMatchers.get(topic).forEach(v -> {
+ any.compareAndSet(false, v.test(message));
+ v.setSatisfied(any.get());
+ });
+ if (!any.get()) {
+ cacheMessage(topic, message);
+ }
+ else {
+ logger.debug("Verified message: {} on topic {}", message, topic);
+ }
+
+ if (!allMatch(topic).get()) {
+ cacheMessage(topic, message);
+ }
+ }
+ }
+
+ static class RabbitTemplateTopicSender implements TestTopicSender {
+
+ private final RabbitTemplate rabbitTemplate;
+
+ private String routingKey = "#";
+
+ RabbitTemplateTopicSender(RabbitTemplate rabbitTemplate) {
+ this.rabbitTemplate = rabbitTemplate;
+ }
+
+ @Override
+ public void send(String topic, P payload) {
+ rabbitTemplate.convertAndSend(topic, routingKey, payload);
+ }
+
+ @Override
+ public void send(String topic, Message message) {
+
+ }
+
+ public void setRoutingKey(@NonNull String routingKey) {
+ this.routingKey = routingKey;
+ }
+ }
+}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppTest.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppTest.java
new file mode 100644
index 00000000..36553dff
--- /dev/null
+++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppTest.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://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.springframework.cloud.stream.app.test.integration.rabbitmq;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Testcontainers
+@ExtendWith(SpringExtension.class)
+@ContextConfiguration(classes = RabbitMQStreamAppContainerTestConfiguration.class)
+public @interface RabbitMQStreamAppTest {
+}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupport.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupport.java
deleted file mode 100644
index 884144fa..00000000
--- a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupport.java
+++ /dev/null
@@ -1,251 +0,0 @@
-/*
- * Copyright 2020-2020 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * https://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.springframework.cloud.stream.app.test.integration.rabbitmq;
-
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Function;
-import java.util.function.Predicate;
-
-import com.github.benmanes.caffeine.cache.Cache;
-import com.github.benmanes.caffeine.cache.Caffeine;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.testcontainers.containers.Network;
-import org.testcontainers.containers.RabbitMQContainer;
-import org.testcontainers.utility.DockerImageName;
-
-import org.springframework.amqp.core.BindingBuilder;
-import org.springframework.amqp.core.MessageProperties;
-import org.springframework.amqp.core.Queue;
-import org.springframework.amqp.core.TopicExchange;
-import org.springframework.amqp.rabbit.annotation.EnableRabbit;
-import org.springframework.amqp.rabbit.annotation.RabbitListener;
-import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
-import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
-import org.springframework.amqp.rabbit.connection.ConnectionFactory;
-import org.springframework.amqp.rabbit.core.RabbitAdmin;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.amqp.support.AmqpHeaders;
-import org.springframework.amqp.support.converter.MessageConversionException;
-import org.springframework.amqp.support.converter.MessageConverter;
-import org.springframework.cloud.stream.app.test.integration.AbstractTestTopicListener;
-import org.springframework.cloud.stream.app.test.integration.MessageMatcher;
-import org.springframework.cloud.stream.app.test.integration.StreamAppContainer;
-import org.springframework.cloud.stream.app.test.integration.StreamApplicationIntegrationTestSupport;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.messaging.Message;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit.jupiter.SpringExtension;
-import org.springframework.util.CollectionUtils;
-
-@ExtendWith(SpringExtension.class)
-@ContextConfiguration(classes = RabbitMQStreamApplicationIntegrationTestSupport.RabbitMQTestConfiguration.class)
-public abstract class RabbitMQStreamApplicationIntegrationTestSupport extends StreamApplicationIntegrationTestSupport {
-
- protected static RabbitMQContainer rabbitmq;
-
- final static String BINDER = "rabbit";
-
- final static Network network = Network.SHARED;
-
- static {
- rabbitmq = new RabbitMQContainer(DockerImageName.parse("rabbitmq:3-management"))
- .withNetwork(network)
- .withExposedPorts(5672, 15672);
- rabbitmq.start();
- }
-
- protected static StreamAppContainer prepackagedRabbitMQContainerFor(String appName, String version) {
- return new RabbitMQStreamAppContainer(prePackagedStreamAppImageName(appName, BINDER, version),
- rabbitmq);
- }
-
- @Configuration
- @EnableRabbit
- static class RabbitMQTestConfiguration {
- public static final String STREAM_APPLICATION_TESTS_GROUP = "stream-application-tests";
-
- @Bean
- public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
- return new RabbitTemplate(connectionFactory);
- }
-
- @Bean
- RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
- return new RabbitAdmin(connectionFactory);
- }
-
- @Bean
- public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
- ConnectionFactory connectionFactory) {
- SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
- factory.setConnectionFactory(connectionFactory);
- factory.setMessageConverter(new MessageConverter() {
- @Override
- public org.springframework.amqp.core.Message toMessage(Object o, MessageProperties messageProperties)
- throws MessageConversionException {
- throw new UnsupportedOperationException("toMessage not implemented.");
- }
-
- @Override
- public Object fromMessage(org.springframework.amqp.core.Message message)
- throws MessageConversionException {
- return new String(message.getBody());
- }
- });
- return factory;
- }
-
- @Bean
- public ConnectionFactory connectionFactory() {
- return new CachingConnectionFactory(localHostAddress(), rabbitmq.getMappedPort(5672));
- }
-
- @Bean
- RabbitMQTestListener rabbitMQTestListener(RabbitAdmin admin) {
- return new RabbitMQTestListener(admin);
- }
-
- static class RabbitMQTestListener extends AbstractTestTopicListener {
-
- public static final int CACHE_TTL_SEC = 120;
-
- private final Cache>> cache = Caffeine.newBuilder()
- .expireAfterWrite(CACHE_TTL_SEC, TimeUnit.SECONDS)
- .build();
-
- static final String STREAM_APPLICATIONS_TEST_QUEUE = "stream-applications-test-queue";
-
- private final RabbitAdmin admin;
-
- private final Queue queue;
-
- private final TopicExchange exchange = new TopicExchange(STREAM_APPLICATIONS_TEST_TOPIC);
-
- RabbitMQTestListener(RabbitAdmin admin) {
- super();
- this.admin = admin;
- this.queue = new Queue(STREAM_APPLICATIONS_TEST_QUEUE);
- admin.declareQueue(queue);
- admin.declareExchange(exchange);
- admin.declareBinding(
- BindingBuilder.bind(queue).to(exchange).with("#"));
- }
-
- @Override
- public AtomicBoolean allMatch(String topic) {
- AtomicBoolean all = super.allMatch(topic);
- if (!all.get()) {
- if (messageMatchers.get(topic) == null || cache.getIfPresent(topic) == null) {
- return all;
- }
- List matchers = messageMatchers.get(topic);
- all.set(true);
- cache.getIfPresent(topic).forEach(message -> matchers.stream().filter(mm -> !mm.isSatisfied())
- .forEach(mm -> all.compareAndSet(true, mm.test(message))));
- }
- return all;
- }
-
- @Override
- public AtomicBoolean matches(String topic, Predicate... predicates) {
- AtomicBoolean matches = super.matches(topic, predicates);
- if (matches.get() || CollectionUtils.isEmpty(cache.getIfPresent(topic))) {
- return matches;
- }
-
- for (Predicate predicate : predicates) {
- MessageMatcher matcher = messageMatcher(topic, predicate)
- .orElse(MessageMatcher.payloadMatcher(o -> false));
- if (messageMatcher(topic, predicate).isPresent()) {
- Set> messages = cache.getIfPresent(topic);
- matches.set(true);
- messages.forEach(message -> {
- if (matches.compareAndSet(true, matcher.test(message))) {
- logger.debug("Matched cached message {} for topic {}", message, topic);
- messages.remove(message);
- return;
- }
- });
- updateCache(topic, messages);
- }
- }
- return matches;
- }
-
- private void updateCache(String topic, Set> messages) {
- if (CollectionUtils.isEmpty(messages)) {
- cache.invalidate(topic);
- }
- else {
- cache.put(topic, messages);
- }
- }
-
- private void cacheMessage(String topic, Message message) {
- if (cache.getIfPresent(topic) == null) {
- cache.put(topic, new HashSet<>());
- }
- Set> messages = cache.getIfPresent(topic);
- if (messages.add(message)) {
- logger.debug("Caching message: {} for topic {}", message, topic);
- }
-
- }
-
- @Override
- protected Function, String> topicForMessage() {
- return message -> (String) message.getHeaders().get(AmqpHeaders.RECEIVED_EXCHANGE);
- }
-
- //@formatter:off
- @RabbitListener(autoStartup = "true", group = STREAM_APPLICATION_TESTS_GROUP,
- queues = {STREAM_APPLICATIONS_TEST_QUEUE})
- //@formatter:on
- @Override
- public void listen(Message message) {
- String topic = topicForMessage().apply(message);
- logger.debug("Received message: {} on topic {}", message, topic);
- if (!messageMatchers.containsKey(topic)) {
- cacheMessage(topic, message);
- return;
- }
-
- logger.debug("Verifying message: {} on topic {}", message, topic);
- AtomicBoolean any = new AtomicBoolean(false);
- messageMatchers.get(topic).forEach(v -> {
- any.compareAndSet(false, v.test(message));
- v.setSatisfied(any.get());
- });
- if (!any.get()) {
- cacheMessage(topic, message);
- }
- else {
- logger.debug("Verified message: {} on topic {}", message, topic);
- }
-
- if (!allMatch(topic).get()) {
- cacheMessage(topic, message);
- }
- }
- }
- }
-}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainerTests.java b/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainerTests.java
new file mode 100644
index 00000000..54bf73dd
--- /dev/null
+++ b/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainerTests.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://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.springframework.cloud.stream.app.test.integration;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+
+import static org.awaitility.Awaitility.await;
+import static org.springframework.cloud.stream.app.test.integration.TestTopicListener.STREAM_APPLICATIONS_TEST_TOPIC;
+
+public abstract class StreamAppContainerTests {
+
+ @Autowired
+ OutputMatcher outputMatcher;
+
+ @Autowired
+ TestTopicSender testTopicSender;
+
+ @AfterEach
+ void reset() {
+ outputMatcher.clearMessageMatchers();
+ }
+
+ @Test
+ void payloadVerifiers() {
+ outputMatcher.addMessageMatcher(MessageMatcher.payloadMatcher((s -> s.equals("hello test1"))));
+ outputMatcher.addMessageMatcher(MessageMatcher.payloadMatcher(s -> s.equals("hello test2")));
+ testTopicSender.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test1");
+ testTopicSender.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test2");
+ await().atMost(Duration.ofSeconds(10))
+ .until(outputMatcher.messagesMatch());
+ }
+
+ @Test
+ void verifierOnTheFly() {
+ testTopicSender.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test3");
+ testTopicSender.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test4");
+ await().atMost(Duration.ofSeconds(30))
+ .until(outputMatcher.payloadMatches(s -> s.equals("hello test3"), s -> s.equals("hello test4")));
+ }
+
+ @Test
+ void verifierOnTheFlyOutOfOrder() {
+ testTopicSender.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test5");
+ testTopicSender.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test6");
+ await().atMost(Duration.ofSeconds(30))
+ .until(outputMatcher.payloadMatches(s -> s.equals("hello test6"), s -> s.equals("hello test5")));
+
+ }
+
+}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainerTests.java b/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainerTests.java
new file mode 100644
index 00000000..013abc29
--- /dev/null
+++ b/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainerTests.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://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.springframework.cloud.stream.app.test.integration.kafka;
+
+import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTests;
+
+@KafkaStreamAppTest
+public class KafkaStreamAppContainerTests extends StreamAppContainerTests {
+}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupportTests.java b/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupportTests.java
deleted file mode 100644
index ccc80882..00000000
--- a/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupportTests.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Copyright 2020-2020 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * https://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.springframework.cloud.stream.app.test.integration.kafka;
-
-import java.time.Duration;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cloud.stream.app.test.integration.MessageMatcher;
-import org.springframework.cloud.stream.app.test.integration.TestTopicListener;
-import org.springframework.kafka.core.KafkaTemplate;
-
-import static org.awaitility.Awaitility.await;
-import static org.springframework.cloud.stream.app.test.integration.AbstractTestTopicListener.STREAM_APPLICATIONS_TEST_TOPIC;
-
-public class KafkaStreamApplicationIntegrationTestSupportTests extends KafkaStreamApplicationIntegrationTestSupport {
-
- @Autowired
- private KafkaTemplate kafkaTemplate;
-
- @Autowired
- private TestTopicListener testTopicListener;
-
- @AfterEach
- void reset() {
- testTopicListener.clearMessageMatchers();
- }
-
- @Test
- void payloadVerifiers() {
- testTopicListener.addMessageMatcher(MessageMatcher.payloadMatcher((s -> s.equals("hello test1"))));
- testTopicListener.addMessageMatcher(MessageMatcher.payloadMatcher(s -> s.equals("hello test2")));
- kafkaTemplate.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test1");
- kafkaTemplate.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test2");
- await().atMost(Duration.ofSeconds(10))
- .until(messagesMatch());
- }
-
- @Test
- void verifierOnTheFly() {
- kafkaTemplate.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test3");
- kafkaTemplate.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test4");
- await().atMost(Duration.ofSeconds(30))
- .until(payloadMatches(s -> s.equals("hello test3"), s -> s.equals("hello test4")));
- }
-
- @Test
- void verifierOnTheFlyOutOfOrder() {
- kafkaTemplate.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test5");
- kafkaTemplate.send(STREAM_APPLICATIONS_TEST_TOPIC, "hello test6");
- await().atMost(Duration.ofSeconds(30))
- .until(payloadMatches(s -> s.equals("hello test6"), s -> s.equals("hello test5")));
-
- }
-}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainerTests.java b/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainerTests.java
new file mode 100644
index 00000000..b5cd8666
--- /dev/null
+++ b/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainerTests.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://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.springframework.cloud.stream.app.test.integration.rabbitmq;
+
+import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTests;
+
+@RabbitMQStreamAppTest
+public class RabbitMQStreamAppContainerTests extends StreamAppContainerTests {
+}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupportTests.java b/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupportTests.java
deleted file mode 100644
index fd803faf..00000000
--- a/applications/stream-applications-core/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupportTests.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * Copyright 2020-2020 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * https://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.springframework.cloud.stream.app.test.integration.rabbitmq;
-
-import java.time.Duration;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
-
-import org.springframework.amqp.rabbit.core.RabbitAdmin;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cloud.stream.app.test.integration.MessageMatcher;
-import org.springframework.cloud.stream.app.test.integration.TestTopicListener;
-
-import static org.awaitility.Awaitility.await;
-import static org.springframework.cloud.stream.app.test.integration.AbstractTestTopicListener.STREAM_APPLICATIONS_TEST_TOPIC;
-
-public class RabbitMQStreamApplicationIntegrationTestSupportTests
- extends RabbitMQStreamApplicationIntegrationTestSupport {
-
- @Autowired
- private RabbitTemplate rabbitTemplate;
-
- @Autowired
- private TestTopicListener testTopicListener;
-
- @Autowired
- RabbitAdmin rabbitAdmin;
-
- @AfterEach
- void reset() {
- testTopicListener.clearMessageMatchers();
- }
-
- @Test
- void multipleVerifiers() {
- testTopicListener.addMessageMatcher(MessageMatcher.payloadMatcher(s -> s.equals("hello test1")));
- testTopicListener.addMessageMatcher(MessageMatcher.payloadMatcher(s -> s.equals("hello test2")));
- rabbitTemplate.convertAndSend(STREAM_APPLICATIONS_TEST_TOPIC, "#", "hello test1");
- rabbitTemplate.convertAndSend(STREAM_APPLICATIONS_TEST_TOPIC, "#", "hello test2");
- await().atMost(Duration.ofSeconds(30))
- .until(messagesMatch());
- }
-
- @Test
- void verifierOnTheFly() {
- rabbitTemplate.convertAndSend(STREAM_APPLICATIONS_TEST_TOPIC, "#", "hello test3");
- rabbitTemplate.convertAndSend(STREAM_APPLICATIONS_TEST_TOPIC, "#", "hello test4");
- await().atMost(Duration.ofSeconds(30))
- .until(payloadMatches(s -> s.equals("hello test3"), s -> s.equals("hello test4")));
- }
-
- @Test
- void verifierOnTheFlyOutOfOrder() {
- rabbitTemplate.convertAndSend(STREAM_APPLICATIONS_TEST_TOPIC, "#", "hello test5");
- rabbitTemplate.convertAndSend(STREAM_APPLICATIONS_TEST_TOPIC, "#", "hello test6");
- await().atMost(Duration.ofSeconds(30))
- .until(payloadMatches(s -> s.equals("hello test6"), s -> s.equals("hello test5")));
- }
-}
diff --git a/applications/stream-applications-core/stream-applications-test-support/src/test/resources/logback-test.xml b/applications/stream-applications-core/stream-applications-test-support/src/test/resources/logback-test.xml
index 4cbc9422..132c60c3 100644
--- a/applications/stream-applications-core/stream-applications-test-support/src/test/resources/logback-test.xml
+++ b/applications/stream-applications-core/stream-applications-test-support/src/test/resources/logback-test.xml
@@ -11,6 +11,7 @@
+