diff --git a/applications/stream-applications-core/README.adoc b/applications/stream-applications-core/README.adoc index 1f493a1c..ec0134d3 100644 --- a/applications/stream-applications-core/README.adoc +++ b/applications/stream-applications-core/README.adoc @@ -1,2 +1,3 @@ -#Core components shared by other projects in the app starters organization +# Core components shared by other projects in the app starters organization This module consists of core dependencies and other common artifacts. + diff --git a/applications/stream-applications-core/common/stream-applications-test-support/README.adoc b/applications/stream-applications-core/common/stream-applications-test-support/README.adoc new file mode 100644 index 00000000..5d58a388 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/README.adoc @@ -0,0 +1,16 @@ +# Stream Applications Test Support + +This module contains common components to support stream applications. + + +## Stream Application Integration Testing + +The `integration` package contains components supporting integration testing stream apps using TestContainers. + +### StreamAppContainer + +An extension + + + + diff --git a/applications/stream-applications-core/common/stream-applications-test-support/pom.xml b/applications/stream-applications-core/common/stream-applications-test-support/pom.xml index e051154a..42309cac 100644 --- a/applications/stream-applications-core/common/stream-applications-test-support/pom.xml +++ b/applications/stream-applications-core/common/stream-applications-test-support/pom.xml @@ -1,58 +1,80 @@ - - stream-applications-core - org.springframework.cloud.stream.app - ${revision} - ../.. - - 4.0.0 + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> + + stream-applications-core + org.springframework.cloud.stream.app + ${revision} + ../.. + + 4.0.0 - stream-applications-test-support + stream-applications-test-support - - - org.springframework.integration - spring-integration-test - - - org.springframework.integration - spring-integration-test-support - - - org.springframework.boot - spring-boot-starter-data-redis - true - - - org.testcontainers - testcontainers - ${test-containers.version} - true - + + + org.springframework.integration + spring-integration-test + + + org.springframework.integration + spring-integration-test-support + + + org.springframework.boot + spring-boot-starter-data-redis + true + + + org.testcontainers + testcontainers + ${test-containers.version} + true + - - org.testcontainers - junit-jupiter - ${test-containers.version} - true - + + org.testcontainers + junit-jupiter + ${test-containers.version} + true + - - org.testcontainers - kafka - ${test-containers.version} - true - + + org.testcontainers + kafka + ${test-containers.version} + true + - - org.awaitility - awaitility - true - + + org.testcontainers + rabbitmq + ${test-containers.version} + true + - + + org.springframework.kafka + spring-kafka + ${spring-kafka.version} + + + org.springframework.amqp + spring-rabbit + ${spring-rabbit.version} + + + + com.github.ben-manes.caffeine + caffeine + + + + org.awaitility + awaitility + + + diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/AbstractTestTopicListener.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/AbstractTestTopicListener.java new file mode 100644 index 00000000..c958faf1 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/AbstractTestTopicListener.java @@ -0,0 +1,151 @@ +/* + * 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.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; +import java.util.function.Predicate; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.springframework.messaging.Message; + +/** + * The base class for + * {@link org.springframework.cloud.stream.app.test.integration.TestTopicListener}s. + * Registers and tests verifiers on incoming messages. Subclasses delegate to this + * listener. + * @author David Turanski + */ +public abstract class AbstractTestTopicListener implements TestTopicListener { + + protected final Map> verifiers = new ConcurrentHashMap<>(); + + protected final Logger logger = LoggerFactory.getLogger(this.getClass()); + + @Override + public boolean addOutputPayloadVerifier(String topic, Predicate verifier) { + Predicate> messagePredicate = message -> verifier.test((T) message.getPayload()); + return addOutputMessageVerifier(topic, messagePredicate); + } + + @Override + public boolean addOutputMessageVerifier(String topic, Predicate> verifier) { + AtomicBoolean isNewTopic = new AtomicBoolean(!verifiers.containsKey(topic)); + + if (isRegisteredOutputVerifier(topic, verifier)) { + return false; + } + + if (isNewTopic.get()) { + logger.trace("Listener is consuming from topic {}", topic); + } + + logger.trace("Setting new output verifier on topic {}", topic); + verifiers.putIfAbsent(STREAM_APPLICATIONS_TEST_TOPIC, new LinkedList<>()); + verifiers.get(STREAM_APPLICATIONS_TEST_TOPIC).add(new Verifier(verifier)); + logger.trace("There are {} output verifiers on topic {}", verifiers.get(STREAM_APPLICATIONS_TEST_TOPIC).size(), + topic); + return true; + } + + private boolean isRegisteredOutputVerifier(String topic, Predicate> newOutputVerifier) { + if (!verifiers.containsKey(topic)) { + return false; + } + AtomicBoolean registered = new AtomicBoolean(false); + verifiers.get(topic).forEach(verifier -> { + if (newOutputVerifier.equals(verifier)) { + logger.debug("This verifier is already registered on topic {}", topic); + registered.set(true); + return; + } + }); + return registered.get(); + } + + @Override + public AtomicBoolean isVerified(String topic) { + AtomicBoolean all = new AtomicBoolean(true); + if (verifiers.containsKey(topic)) { + verifiers.get(topic).forEach(v -> all.compareAndSet(true, v.isSatisfied())); + } + + logger.trace("Verified topic {} is {}", topic, all.get()); + return all; + } + + @Override + public void clearOutputVerifiers() { + verifiers.clear(); + } + + @Override + public void resetOutputVerifiers() { + verifiers.values().forEach((List l) -> l.forEach(v -> v.setSatisfied(false))); + } + + protected abstract Function, String> topicForMessage(); + + @Override + public void listen(Message message) { + String topic = topicForMessage().apply(message); + logger.debug("Received message: {} on topic {}", message, topic); + if (!verifiers.containsKey(topic)) { + return; + } + + logger.trace("Verifying message: {} on topic {}", message, topic); + AtomicBoolean any = new AtomicBoolean(false); + verifiers.get(topic).forEach(v -> { + any.compareAndSet(false, v.test(message)); + v.setSatisfied(any.get()); + }); + if (any.get()) { + logger.debug("Verified message: {} on topic {}", message, topic); + } + } + + protected final static class Verifier implements Predicate> { + private final Predicate> predicate; + + private final AtomicBoolean satisfied; + + private Verifier(Predicate> predicate) { + this.predicate = predicate; + this.satisfied = new AtomicBoolean(false); + } + + @Override + public boolean test(Message message) { + return predicate.test(message); + } + + public void setSatisfied(boolean value) { + this.satisfied.compareAndSet(false, value); + } + + public boolean isSatisfied() { + return this.satisfied.get(); + } + } +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java index 633fd0e9..2271b207 100644 --- a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java @@ -16,11 +16,9 @@ package org.springframework.cloud.stream.app.test.integration; -import java.util.LinkedList; -import java.util.List; -import java.util.Optional; import java.util.concurrent.Callable; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.regex.Pattern; @@ -28,65 +26,84 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.testcontainers.containers.output.OutputFrame; +import org.springframework.util.Assert; + /** - * Utility for matching test container log contents. + * Utility for matching test container log contents using Awaitility. Example: + * {@code await().until(logMatcher.matches();} * @author David Turanski */ public class LogMatcher implements Consumer { private static Logger logger = LoggerFactory.getLogger(LogMatcher.class); - private List> listeners = new LinkedList<>(); + protected AtomicBoolean matched = new AtomicBoolean(); - public Callable verifies(Consumer consumer) { - LogListener logListener = new LogListener(); - consumer.accept(logListener); - logListener.runnable.ifPresent(runnable -> runnable.run()); - listeners.add(logListener); - return () -> logListener.matches().get(); + private Pattern pattern; + + private LogMatcher() { + } + + public Callable matches() { + return () -> matched.get(); + } + + private LogMatcher(Pattern pattern) { + this.pattern = pattern; + } + + public static LogMatcher contains(String string) { + return LogMatcher.matchesRegex(".*" + string + ".*"); + } + + public static LogMatcher endsWith(String string) { + return LogMatcher.matchesRegex(".*" + string); + } + + public static LogMatcher matchesRegex(String regex) { + return new LogMatcher(Pattern.compile(regex)); + } + + public LogMatcher times(int times) { + return new CountingLogMatcher(this.pattern, times); } @Override public void accept(OutputFrame outputFrame) { - listeners.forEach(m -> m.accept(outputFrame.getUtf8String())); + synchronized (matched) { + if (!matched.get()) { + String str = outputFrame.getUtf8String().trim(); + logger.trace("matching {} using pattern {}", str, pattern.pattern()); + if (pattern.matcher(str).matches()) { + matched.set(true); + logger.debug(" MATCHED {}", str); + } + } + } } - public class LogListener implements Consumer { - private AtomicBoolean matched = new AtomicBoolean(); + public final static class CountingLogMatcher extends LogMatcher { - private Optional runnable = Optional.empty(); + private final AtomicInteger count = new AtomicInteger(); - private Pattern pattern; + private CountingLogMatcher(Pattern pattern, int count) { + super(pattern); + Assert.isTrue(count >= 1, "'count' must be greater than 0"); + this.count.set(count); + } @Override - public void accept(String s) { - logger.trace(this + "matching " + s.trim() + " using pattern " + pattern.pattern()); - if (pattern.matcher(s.trim()).matches()) { - logger.debug(" MATCHED " + s.trim()); - matched.set(true); - listeners.remove(this); + public void accept(OutputFrame outputFrame) { + if (count.get() > 0) { + super.accept(outputFrame); + if (matched.compareAndSet(true, false)) { + count.decrementAndGet(); + } } } - public LogListener contains(String string) { - return matchesRegex(".*" + string + ".*"); - } - - public LogListener endsWith(String string) { - return matchesRegex(".*" + string); - } - - public LogListener matchesRegex(String regex) { - this.pattern = Pattern.compile(regex); - return this; - } - - public LogListener when(Runnable runnable) { - this.runnable = Optional.of(runnable); - return this; - } - - public AtomicBoolean matches() { - return matched; + @Override + public Callable matches() { + return () -> count.get() == 0; } } } diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainer.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainer.java new file mode 100644 index 00000000..49019028 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainer.java @@ -0,0 +1,116 @@ +/* + * 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.util.UUID; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.utility.DockerImageName; + +import org.springframework.util.Assert; + +import static org.springframework.cloud.stream.app.test.integration.AppLog.appLog; + +/** + * Extends {@link org.testcontainers.containers.GenericContainer} to support dockerized + * Spring Cloud Stream applications. Currently this only supports apps with single I/O + * destinations. This configures standard input and output destination bindings. + * @author David Turanski + */ +public abstract class StreamAppContainer extends GenericContainer { + + protected Logger logger = LoggerFactory.getLogger(this.getClass()); + + protected final GenericContainer messageBrokerContainer; + + private String inputDestination; + + private String outputDestination; + + /** + * Create a TestContainer with standard Spring Cloud Stream input and output bindings. For + * function based apps, you need to alias the function endpoints to "input" and "output". + * The output destination is set to + * {@code TestTopicListener.STREAM_APPLICATIONS_TEST_TOPIC}. The input destination is a + * random value. + * @param imageName the name of the image. + * @param messageBrokerContainer a message broker TestContainer. Typically, it is a + * singleton TestContainer created in a static initializer. + */ + public StreamAppContainer(String imageName, GenericContainer messageBrokerContainer) { + super(DockerImageName.parse(imageName)); + Assert.notNull(messageBrokerContainer, "A Message broker container is required."); + Assert.isTrue(messageBrokerContainer.isRunning(), "Message broker container must be started first."); + this.messageBrokerContainer = messageBrokerContainer; + this.withNetwork(messageBrokerContainer.getNetwork()).dependsOn(this.messageBrokerContainer) + .withOutputDestination(TestTopicListener.STREAM_APPLICATIONS_TEST_TOPIC) + .withInputDestination(TestTopicListener.STREAM_APPLICATIONS_TEST_TOPIC + "_IN_" + + UUID.randomUUID().toString().substring(0, 8)) + .withBinderProperties(); + if (logger.isDebugEnabled()) { + this.log(); + } + } + + /** + * @return the input destination. + */ + public String getInputDestination() { + return inputDestination; + } + + /** + * @return the output destination. + */ + public String getOutputDestination() { + return outputDestination; + } + + /** + * Enable container logging. This is invoked if the class logger is set to DEBUG. + * @return the instance. + */ + public StreamAppContainer log() { + this.withLogConsumer(appLog(this.getImage().get())); + return this; + } + + /** + * Assign a destination name to the standard input. + * @param destination the destination name. + * @return + */ + public StreamAppContainer withInputDestination(String destination) { + Assert.hasText(destination, "'destination' is required."); + withEnv("SPRING_CLOUD_STREAM_BINDINGS_INPUT_DESTINATION", destination); + withEnv("SPRING_CLOUD_STREAM_BINDINGS_INPUT_GROUP", TestTopicListener.STREAM_APPLICATIONS_TEST_TOPIC + "_GROUP_" + + UUID.randomUUID().toString().substring(0, 8)); + inputDestination = destination; + return this; + } + + private StreamAppContainer withOutputDestination(String destination) { + Assert.hasText(destination, "'destination' is required."); + withEnv("SPRING_CLOUD_STREAM_BINDINGS_OUTPUT_DESTINATION", destination); + outputDestination = destination; + return this; + } + + protected abstract StreamAppContainer withBinderProperties(); +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamIApplicationIntegrationTestSupport.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApplicationIntegrationTestSupport.java similarity index 57% rename from applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamIApplicationIntegrationTestSupport.java rename to applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApplicationIntegrationTestSupport.java index 5d3b2063..b6774ee1 100644 --- a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamIApplicationIntegrationTestSupport.java +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApplicationIntegrationTestSupport.java @@ -20,10 +20,15 @@ import java.io.File; import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; +import java.util.concurrent.Callable; +import java.util.function.Predicate; import org.testcontainers.junit.jupiter.Testcontainers; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.core.io.ClassPathResource; +import org.springframework.messaging.Message; +import org.springframework.stereotype.Component; import org.springframework.util.SocketUtils; /** @@ -31,9 +36,19 @@ import org.springframework.util.SocketUtils; * @author David Turanski */ @Testcontainers -public abstract class StreamIApplicationIntegrationTestSupport { +@Component +public abstract class StreamApplicationIntegrationTestSupport { - protected static String localHostAddress() { + protected static final String DOCKER_ORG = "springcloudstream"; + + @Autowired + private AbstractTestTopicListener testListener; + + protected static String prePackagedStreamAppImageName(String appName, String binderName, String version) { + return DOCKER_ORG + "/" + appName + "-" + binderName + ":" + version; + } + + public static String localHostAddress() { try { return InetAddress.getLocalHost().getHostAddress(); } @@ -55,4 +70,17 @@ public abstract class StreamIApplicationIntegrationTestSupport { return SocketUtils.findAvailableTcpPort(10000, 20000); } + protected Callable verifyOutputMessages() { + return () -> testListener.isVerified().get(); + } + + protected

Callable verifyOutputPayload(Predicate

outputVerifier) { + testListener.addOutputPayloadVerifier(outputVerifier); + return () -> testListener.isVerified().get(); + } + + protected Callable verifyOutputMessage(Predicate> outputVerifier) { + testListener.addOutputMessageVerifier(outputVerifier); + return () -> testListener.isVerified().get(); + } } diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApps.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApps.java index e9953f39..646baaa2 100644 --- a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApps.java +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApps.java @@ -30,8 +30,6 @@ import org.testcontainers.lifecycle.Startable; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; -import static org.springframework.cloud.stream.app.test.integration.AppLog.appLog; - public abstract class StreamApps implements AutoCloseable, Startable { protected Logger logger = LoggerFactory.getLogger(this.getClass()); @@ -63,25 +61,7 @@ public abstract class StreamApps implements AutoCloseable, Startable { public void start() { if (logger.isDebugEnabled()) { - logger.debug("Starting apps..."); - - logger.debug("Source container environment:"); - sourceContainer().getEnv().forEach((Consumer) env -> logger.debug(env)); - sourceContainer().withLogConsumer(appLog(sourceContainer().getImage().get())); - - if (!CollectionUtils.isEmpty(processorContainers)) { - logger.debug("\nProcessor containers environment:"); - processorContainers().forEach(container -> { - logger.debug("Processor container environment:"); - container.getEnv().forEach((Consumer) env -> logger.debug(env)); - container.withLogConsumer(appLog(container.getImage().get())); - - }); - } - - logger.debug("\nSink container environment:"); - sinkContainer().getEnv().forEach((Consumer) env -> logger.debug(env)); - sinkContainer().withLogConsumer(appLog(sinkContainer().getImage().get())); + logDebugInfo(); } sinkContainer.start(); @@ -95,7 +75,23 @@ public abstract class StreamApps implements AutoCloseable, Startable { sourceContainer.stop(); } - public static abstract class Builder { + private void logDebugInfo() { + logger.debug("Starting apps..."); + logger.debug("Source container environment for {} :", sourceContainer().getImage().get()); + sourceContainer().getEnv().forEach((Consumer) env -> logger.debug(env)); + + if (!CollectionUtils.isEmpty(processorContainers)) { + logger.debug("\nProcessor containers environment:"); + processorContainers().forEach(container -> { + logger.debug("Processor container environment for {}", container.getImage().get()); + container.getEnv().forEach((Consumer) env -> logger.debug(env)); + }); + } + logger.debug("\nSink container environment for {} :", sinkContainer().getImage().get()); + sinkContainer().getEnv().forEach((Consumer) env -> logger.debug(env)); + } + + public static abstract class Builder { private final String streamName; private GenericContainer source; @@ -129,15 +125,17 @@ public abstract class StreamApps implements AutoCloseable, Startable { return this; } - public StreamApps build() { + public S build() { Assert.notNull(source, "A Source container is required."); Assert.notNull(sink, "A Sink container is required."); - return streamAppsInstance(setupSourceContainer(), setupProcessorContainers(), setupSinkContainer()); + return doBuild(setupSourceContainer(), setupProcessorContainers(), setupSinkContainer()); } - protected abstract StreamApps streamAppsInstance(GenericContainer sourceContainer, + protected abstract Map binderProperties(); + + protected abstract S doBuild(GenericContainer sourceContainer, List processorContainers, GenericContainer sinkContainer); private GenericContainer setupSourceContainer() { @@ -178,7 +176,5 @@ public abstract class StreamApps implements AutoCloseable, Startable { return (CollectionUtils.isEmpty(processors) || processors.size() <= 1) ? streamName : "processor_" + (processors.size() - 1); } - - protected abstract Map binderProperties(); } } diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/TestTopicListener.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/TestTopicListener.java new file mode 100644 index 00000000..0a480a37 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/TestTopicListener.java @@ -0,0 +1,104 @@ +/* + * 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.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Predicate; + +import org.springframework.messaging.Message; + +/** + * The contract for TestTopicListener implementations. + * + * @author David Turanski + */ +public interface TestTopicListener { + + /** + * Default Output Destination. + */ + String STREAM_APPLICATIONS_TEST_TOPIC = "stream-applications-test"; + + /** + * Register a Message payload verifier verifier on the given destination. + * @param topic the destination for the output Message. + * @param outputVerifier a {code Predicate} to test the payload. + * @param

the expected payload type + * @return true if it is registered, false if it is already registered. + */ +

boolean addOutputPayloadVerifier(String topic, Predicate

outputVerifier); + + /** + * Register a Message payload verifier on the default destination. + * @param outputVerifier a {code Predicate} to test the payload. + * @param

the expected payload type + * @return true if it is registered, false if it is already registered. + */ + default

boolean addOutputPayloadVerifier(Predicate

outputVerifier) { + return addOutputPayloadVerifier(STREAM_APPLICATIONS_TEST_TOPIC, outputVerifier); + } + + /** + * Register a {@link Message} verifier on the given destination. + * @param topic the destination for the output Message. + * @param outputVerifier a {code Predicate} to test the payload. + * @return true if it is registered, false if it is already registered. + */ + boolean addOutputMessageVerifier(String topic, Predicate> outputVerifier); + + /** + * Register a Message payload verifier on the default destination. + * @param outputVerifier a {code Predicate} to test the payload.e + * @return true if it is registered, false if it is already registered. + */ + default boolean addOutputMessageVerifier(Predicate> outputVerifier) { + return addOutputMessageVerifier(STREAM_APPLICATIONS_TEST_TOPIC, outputVerifier); + } + + /** + * Remove all verifiers. + */ + void clearOutputVerifiers(); + + /** + * Set all verifiers to the initial state. + */ + void resetOutputVerifiers(); + + /** + * A method that may be polled to wait for all verifiers on a given destination to be + * satisfied. + * @param topic the destination. + * @return true if all verifiers are satisfied. + */ + AtomicBoolean isVerified(String topic); + + /** + * A method that may be polled to wait for all verifiers on the default destination to be + * satisfied. + * @return true if all verifiers are satisfied. + */ + default AtomicBoolean isVerified() { + return isVerified(STREAM_APPLICATIONS_TEST_TOPIC); + } + + /** + * A message listener to a topic and tests all verifiers on an incoming Message. + * @param message the Message. + */ + void listen(Message message); +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/AbstractKafkaStreamApplicationIntegrationTests.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainer.java similarity index 51% rename from applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/AbstractKafkaStreamApplicationIntegrationTests.java rename to applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainer.java index cfe456e5..364c3b38 100644 --- a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/AbstractKafkaStreamApplicationIntegrationTests.java +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainer.java @@ -16,27 +16,29 @@ package org.springframework.cloud.stream.app.test.integration.kafka; -import org.testcontainers.containers.KafkaContainer; -import org.testcontainers.containers.Network; -import org.testcontainers.junit.jupiter.Testcontainers; -import org.testcontainers.utility.DockerImageName; +import org.testcontainers.containers.GenericContainer; -import org.springframework.cloud.stream.app.test.integration.StreamIApplicationIntegrationTestSupport; +import org.springframework.cloud.stream.app.test.integration.StreamAppContainer; /** - * Base class for stream application integration testing with Test Containers and Kafka - * binder + * An implementation of + * {@link org.springframework.cloud.stream.app.test.integration.StreamAppContainer} for + * kafka. This provides the required broker connection properties. */ -@Testcontainers -public abstract class AbstractKafkaStreamApplicationIntegrationTests extends StreamIApplicationIntegrationTestSupport { +public class KafkaStreamAppContainer extends StreamAppContainer { - final static Network network = Network.SHARED; + /** + * @param imageName the image name. + * @param kafka a running kafka TestContainer instance. + */ + public KafkaStreamAppContainer(String imageName, GenericContainer kafka) { + super(imageName, kafka); + } - protected final static KafkaContainer kafka = new KafkaContainer( - DockerImageName.parse("confluentinc/cp-kafka:5.5.1")) - .withNetwork(network); - - static { - kafka.start(); + @Override + protected StreamAppContainer withBinderProperties() { + this.withEnv("SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS", + messageBrokerContainer.getNetworkAliases().get(0) + ":9092"); + return this; } } diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupport.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupport.java new file mode 100644 index 00000000..9a6aabbc --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupport.java @@ -0,0 +1,205 @@ +/* + * 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 java.util.function.Predicate; + +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.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 addOutputMessageVerifier(String topic, Predicate> verifier) { + boolean added = super.addOutputMessageVerifier(topic, verifier); + 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/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApps.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApps.java index 3e44ea1e..0cca6ccc 100644 --- a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApps.java +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApps.java @@ -31,11 +31,11 @@ public class KafkaStreamApps extends StreamApps { super(sourceContainer, processorContainers, sinkContainer); } - public static Builder kafkaStreamApps(String streamName, GenericContainer messageBrokerContainer) { + public static Builder kafkaStreamApps(String streamName, GenericContainer messageBrokerContainer) { return new KafkaBuilder(streamName, messageBrokerContainer); } - public static final class KafkaBuilder extends Builder { + public static final class KafkaBuilder extends Builder { protected KafkaBuilder(String streamName, GenericContainer messageBrokerContainer) { super(streamName, messageBrokerContainer); @@ -47,7 +47,7 @@ public class KafkaStreamApps extends StreamApps { } @Override - protected StreamApps streamAppsInstance(GenericContainer sourceContainer, + protected KafkaStreamApps doBuild(GenericContainer sourceContainer, List processorContainers, GenericContainer sinkContainer) { return new KafkaStreamApps(sourceContainer, processorContainers, sinkContainer); } diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainer.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainer.java new file mode 100644 index 00000000..82c7bdec --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainer.java @@ -0,0 +1,44 @@ +/* + * 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.GenericContainer; + +import org.springframework.cloud.stream.app.test.integration.StreamAppContainer; + +/** + * An implementation of + * {@link org.springframework.cloud.stream.app.test.integration.StreamAppContainer} for + * rabbitMQ. This provides the required broker connection properties. + */ +public class RabbitMQStreamAppContainer extends StreamAppContainer { + + /** + * @param imageName the image name. + * @param rabbitmq a running rabbitMQ TestContainer instance. + */ + public RabbitMQStreamAppContainer(String imageName, GenericContainer rabbitmq) { + super(imageName, rabbitmq); + } + + @Override + protected StreamAppContainer withBinderProperties() { + this.withEnv("SPRING_RABBITMQ_HOST", messageBrokerContainer.getNetworkAliases().get(0).toString()) + .withEnv("SPRING_RABBITMQ_PORT", "5672"); + return this; + } +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupport.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupport.java new file mode 100644 index 00000000..93de7696 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupport.java @@ -0,0 +1,218 @@ +/* + * 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.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; + +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.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; + +@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(); + + private 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 isVerified(String topic) { + AtomicBoolean all = super.isVerified(topic); + if (cache.getIfPresent(topic) != null) { + if (!all.get()) { + all.set(true); + logger.debug("Verifying cached messages for topic {}", topic); + cache.getIfPresent(topic).forEach(m -> verifiers.get(topic).forEach(v -> { + if (!v.isSatisfied()) { + v.setSatisfied(v.test(m)); + all.compareAndSet(true, v.isSatisfied()); + if (v.isSatisfied()) { + cache.invalidate(m); + } + } + })); + } + } + return all; + } + + 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 (!verifiers.containsKey(topic)) { + cacheMessage(topic, message); + return; + } + + logger.debug("Verifying message: {} on topic {}", message, topic); + AtomicBoolean any = new AtomicBoolean(false); + verifiers.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 (!isVerified(topic).get()) { + cacheMessage(topic, message); + } + } + } + } +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApps.java b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApps.java new file mode 100644 index 00000000..a9c751ee --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApps.java @@ -0,0 +1,60 @@ +/* + * 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.List; +import java.util.Map; + +import org.testcontainers.containers.GenericContainer; + +import org.springframework.cloud.stream.app.test.integration.StreamApps; + +import static org.springframework.cloud.stream.app.test.integration.FluentMap.fluentMap; + +public class RabbitMQStreamApps extends StreamApps { + + protected RabbitMQStreamApps(GenericContainer sourceContainer, List processorContainers, + GenericContainer sinkContainer) { + super(sourceContainer, processorContainers, sinkContainer); + } + + public static Builder rabbitMQStreamApps(String streamName, + GenericContainer messageBrokerContainer) { + return new RabbitMQBuilder(streamName, messageBrokerContainer); + } + + public static final class RabbitMQBuilder extends Builder { + + protected RabbitMQBuilder(String streamName, GenericContainer messageBrokerContainer) { + super(streamName, messageBrokerContainer); + } + + protected Map binderProperties() { + + return fluentMap() + .withEntry("SPRING_RABBITMQ_HOST", + messageBrokerContainer.getNetworkAliases().get(0)) + .withEntry("SPRING_RABBITMQ_PORT", "5672"); + } + + @Override + protected RabbitMQStreamApps doBuild(GenericContainer sourceContainer, + List processorContainers, GenericContainer sinkContainer) { + return new RabbitMQStreamApps(sourceContainer, processorContainers, sinkContainer); + } + } +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupportTests.java b/applications/stream-applications-core/common/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupportTests.java new file mode 100644 index 00000000..062ce8b3 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupportTests.java @@ -0,0 +1,74 @@ +/* + * 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.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.clearOutputVerifiers(); + } + + @Test + void payloadVerifiers() { + testTopicListener.addOutputPayloadVerifier((s -> s.equals("hello test1"))); + testTopicListener.addOutputPayloadVerifier((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(verifyOutputMessages()); + } + + @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(verifyOutputPayload((s -> s.equals("hello test4")))); + await().atMost(Duration.ofSeconds(30)) + .until(verifyOutputPayload((s -> s.equals("hello test3")))); + + } + + @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(verifyOutputPayload((s -> s.equals("hello test6")))); + await().atMost(Duration.ofSeconds(30)) + .until(verifyOutputPayload((s -> s.equals("hello test5")))); + } +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupportTests.java b/applications/stream-applications-core/common/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupportTests.java new file mode 100644 index 00000000..8c772e5f --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupportTests.java @@ -0,0 +1,74 @@ +/* + * 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.RabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +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; + + @AfterEach + void reset() { + testTopicListener.clearOutputVerifiers(); + } + + @Test + void multipleVerifiers() { + testTopicListener.addOutputPayloadVerifier((s -> s.equals("hello test1"))); + testTopicListener.addOutputPayloadVerifier((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(verifyOutputMessages()); + } + + @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(verifyOutputPayload((s -> s.equals("hello test3")))); + await().atMost(Duration.ofSeconds(30)) + .until(verifyOutputPayload((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(verifyOutputPayload((s -> s.equals("hello test6")))); + await().atMost(Duration.ofSeconds(30)) + .until(verifyOutputPayload((s -> s.equals("hello test5")))); + } +} diff --git a/applications/stream-applications-core/common/stream-applications-test-support/src/test/resources/logback-test.xml b/applications/stream-applications-core/common/stream-applications-test-support/src/test/resources/logback-test.xml new file mode 100644 index 00000000..4cbc9422 --- /dev/null +++ b/applications/stream-applications-core/common/stream-applications-test-support/src/test/resources/logback-test.xml @@ -0,0 +1,17 @@ + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n + + + + + + + + + + + + + \ No newline at end of file diff --git a/stream-applications-build/pom.xml b/stream-applications-build/pom.xml index 24394068..61f1c3f8 100644 --- a/stream-applications-build/pom.xml +++ b/stream-applications-build/pom.xml @@ -1,351 +1,356 @@ - - 4.0.0 + + 4.0.0 - org.springframework.cloud.stream.app - stream-applications-build - 1.0.0-SNAPSHOT - stream-applications-build - Common Parent for Functions and Applications - pom + org.springframework.cloud.stream.app + stream-applications-build + 1.0.0-SNAPSHOT + stream-applications-build + Common Parent for Functions and Applications + pom - - 1.8 - 3.1.1 - 3.2.1 - 2.22.2 - UTF-8 - UTF-8 - ${java.version} - ${java.version} - 3.0.1 - 3.1.0 - false - true - true - true - 8.29 - https://raw.githubusercontent.com/spring-cloud/stream-applications/master/etc/checkstyle - - ${checkstyle.location}/checkstyle-suppressions.xml - - - ${checkstyle.location}/nohttp-checkstyle.xml - - - ${checkstyle.location}/checkstyle-suppressions.xml - - 0.0.2.RELEASE - true - 0.0.7 - 2.3.4.RELEASE - 3.0.9.RELEASE - 5.3.3.BUILD-SNAPSHOT - 1.15.0-rc2 - 1.2.5 - 2.22.2 - + + 1.8 + 3.1.1 + 3.2.1 + 2.22.2 + UTF-8 + UTF-8 + ${java.version} + ${java.version} + 3.0.1 + 3.1.0 + false + true + true + true + 8.29 + https://raw.githubusercontent.com/spring-cloud/stream-applications/master/etc/checkstyle + + + ${checkstyle.location}/checkstyle-suppressions.xml + + + ${checkstyle.location}/nohttp-checkstyle.xml + + + ${checkstyle.location}/checkstyle-suppressions.xml + + 0.0.2.RELEASE + true + 0.0.7 + 2.3.4.RELEASE + 2.6.0-M1 + 2.2.11.RELEASE + 3.0.9.RELEASE + 5.3.3.BUILD-SNAPSHOT + 1.15.0-rc2 + 1.2.5 + 2.22.2 + - - - - - org.springframework.integration - spring-integration-bom - ${spring-integration-dependencies.version} - import - pom - - - org.springframework.cloud - spring-cloud-function-dependencies - ${spring-cloud-function.version} - import - pom - - - org.springframework.boot - spring-boot-dependencies - ${spring-boot.version} - pom - import - - - + + + + + org.springframework.integration + spring-integration-bom + ${spring-integration-dependencies.version} + import + pom + + + org.springframework.cloud + spring-cloud-function-dependencies + ${spring-cloud-function.version} + import + pom + + + org.springframework.boot + spring-boot-dependencies + ${spring-boot.version} + pom + import + + + - - - - org.codehaus.mojo - flatten-maven-plugin - ${maven-flatten-plugin.version} - - true - resolveCiFriendliesOnly - - - - flatten - process-resources - - flatten - - - - flatten.clean - clean - - clean - - - - + + + + org.codehaus.mojo + flatten-maven-plugin + ${maven-flatten-plugin.version} + + true + resolveCiFriendliesOnly + + + + flatten + process-resources + + flatten + + + + flatten.clean + clean + + clean + + + + - - maven-javadoc-plugin - ${maven-javadoc-plugin.version} - - - javadoc - package - - jar - - - - - true - - + + maven-javadoc-plugin + ${maven-javadoc-plugin.version} + + + javadoc + package + + jar + + + + + true + + - - maven-source-plugin - ${maven-source-plugin.version} - - - attach-sources - package - - jar - - - - + + maven-source-plugin + ${maven-source-plugin.version} + + + attach-sources + package + + jar + + + + - - org.apache.maven.plugins - maven-surefire-plugin - ${maven-surefire-plugin.version} - - - **/*Tests.java - **/*Test.java - - - **/Abstract*.java - - - + + org.apache.maven.plugins + maven-surefire-plugin + ${maven-surefire-plugin.version} + + + **/*Tests.java + **/*Test.java + + + **/Abstract*.java + + + - - org.apache.maven.plugins - maven-checkstyle-plugin - ${maven-checkstyle-plugin.version} - - - com.puppycrawl.tools - checkstyle - ${puppycrawl-tools-checkstyle.version} - - - io.spring.javaformat - spring-javaformat-checkstyle - ${spring-javaformat-checkstyle.version} - - - io.spring.nohttp - nohttp-checkstyle - ${nohttp-checkstyle.version} - - - - - checkstyle-validation - validate - true - - ${disable.checks} - ${checkstyle.location}/checkstyle.xml - ${checkstyle.location}/checkstyle-header.txt - - checkstyle.build.directory=${project.build.directory} - checkstyle.suppressions.file=${checkstyle.suppressions.file} - checkstyle.additional.suppressions.file=${checkstyle.additional.suppressions.file} - - true - - ${maven-checkstyle-plugin.includeTestSourceDirectory} - - ${maven-checkstyle-plugin.failsOnError} - - - ${maven-checkstyle-plugin.failOnViolation} - - - - check - - - - no-http-checkstyle-validation - validate - true - - ${disable.nohttp.checks} - ${checkstyle.nohttp.file} - **/* - **/.idea/**/*,**/.git/**/*,**/target/**/*,**/*.log - ./ - - - check - - - - + + org.apache.maven.plugins + maven-checkstyle-plugin + ${maven-checkstyle-plugin.version} + + + com.puppycrawl.tools + checkstyle + ${puppycrawl-tools-checkstyle.version} + + + io.spring.javaformat + spring-javaformat-checkstyle + ${spring-javaformat-checkstyle.version} + + + io.spring.nohttp + nohttp-checkstyle + ${nohttp-checkstyle.version} + + + + + checkstyle-validation + validate + true + + ${disable.checks} + ${checkstyle.location}/checkstyle.xml + ${checkstyle.location}/checkstyle-header.txt + + checkstyle.build.directory=${project.build.directory} + checkstyle.suppressions.file=${checkstyle.suppressions.file} + checkstyle.additional.suppressions.file=${checkstyle.additional.suppressions.file} + + true + + ${maven-checkstyle-plugin.includeTestSourceDirectory} + + ${maven-checkstyle-plugin.failsOnError} + + + ${maven-checkstyle-plugin.failOnViolation} + + + + check + + + + no-http-checkstyle-validation + validate + true + + ${disable.nohttp.checks} + ${checkstyle.nohttp.file} + **/* + **/.idea/**/*,**/.git/**/*,**/target/**/*,**/*.log + ./ + + + check + + + + - + - + - + - - Apache License, Version 2.0 - http://www.apache.org/licenses/LICENSE-2.0 - Copyright 2014-2020 the original author or authors. + + Apache License, Version 2.0 + http://www.apache.org/licenses/LICENSE-2.0 + Copyright 2014-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 + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at - http://www.apache.org/licenses/LICENSE-2.0 + 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. + 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. - + See the License for the specific language governing permissions and + limitations under the License. + + - + - - scm:git:git://github.com/pivotal/java-functions.git - scm:git:ssh://git@github.com/pivotal/java-functions.git - https://github.com/pivotal/java-functions - + + scm:git:git://github.com/pivotal/java-functions.git + scm:git:ssh://git@github.com/pivotal/java-functions.git + https://github.com/pivotal/java-functions + - + - - repo.spring.io - Spring Release Repository - https://repo.spring.io/libs-release-local - + + repo.spring.io + Spring Release Repository + https://repo.spring.io/libs-release-local + - - repo.spring.io - Spring Snapshot Repository - https://repo.spring.io/libs-snapshot-local - + + repo.spring.io + Spring Snapshot Repository + https://repo.spring.io/libs-snapshot-local + - + - + - - spring-milestones - Spring Milestones - https://repo.spring.io/milestone - + + spring-milestones + Spring Milestones + https://repo.spring.io/milestone + - - spring-snapshots - Spring Snapshots - https://repo.spring.io/snapshot - - true - - + + spring-snapshots + Spring Snapshots + https://repo.spring.io/snapshot + + true + + - + - + - - spring-milestones - Spring Milestones - https://repo.spring.io/milestone - + + spring-milestones + Spring Milestones + https://repo.spring.io/milestone + - - spring-snapshots - Spring Snapshots - https://repo.spring.io/snapshot - - true - - + + spring-snapshots + Spring Snapshots + https://repo.spring.io/snapshot + + true + + - + - + - - milestone - - - repo.spring.io - Spring Milestone Repository - https://repo.spring.io/libs-milestone-local - - - + + milestone + + + repo.spring.io + Spring Milestone Repository + https://repo.spring.io/libs-milestone-local + + + - - central - - - - maven-gpg-plugin - - - sign-artifacts - verify - - sign - - - - - - - - - sonatype-nexus-staging - Nexus Release Repository - https://oss.sonatype.org/service/local/staging/deploy/maven2/ - - - sonatype-nexus-snapshots - Sonatype Nexus Snapshots - https://oss.sonatype.org/content/repositories/snapshots/ - - - + + central + + + + maven-gpg-plugin + + + sign-artifacts + verify + + sign + + + + + + + + + sonatype-nexus-staging + Nexus Release Repository + https://oss.sonatype.org/service/local/staging/deploy/maven2/ + + + sonatype-nexus-snapshots + Sonatype Nexus Snapshots + https://oss.sonatype.org/content/repositories/snapshots/ + + + - +