SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Major refactor to use single default topic for TestTopicListener and add RabbitMQ support

Browse Source
This commit is contained in:
David Turanski
2020-10-19 18:17:00 -04:00
parent 1118a2a7cf
commit b60df2c827
19 changed files with 1603 additions and 453 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
applications/stream-applications-core/README.adoc
View File

@@ -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.

16
applications/stream-applications-core/common/stream-applications-test-support/README.adoc Normal file
View File

@@ -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

116
applications/stream-applications-core/common/stream-applications-test-support/pom.xml
View File

@@ -1,58 +1,80 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>stream-applications-core</artifactId>
<groupId>org.springframework.cloud.stream.app</groupId>
<version>${revision}</version>
<relativePath>../..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>stream-applications-core</artifactId>
<groupId>org.springframework.cloud.stream.app</groupId>
<version>${revision}</version>
<relativePath>../..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>stream-applications-test-support</artifactId>
<artifactId>stream-applications-test-support</artifactId>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-test</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-test-support</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<version>${test-containers.version}</version>
<optional>true</optional>
</dependency>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-test</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-test-support</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<version>${test-containers.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<version>${test-containers.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<version>${test-containers.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>kafka</artifactId>
<version>${test-containers.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>kafka</artifactId>
<version>${test-containers.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>rabbitmq</artifactId>
<version>${test-containers.version}</version>
<optional>true</optional>
</dependency>
</dependencies>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>${spring-kafka.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>${spring-rabbit.version}</version>
</dependency>
<dependency>
<groupId>com.github.ben-manes.caffeine</groupId>
<artifactId>caffeine</artifactId>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
</dependency>
</dependencies>
</project>

151
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/AbstractTestTopicListener.java Normal file
View File

@@ -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<String, List<Verifier>> verifiers = new ConcurrentHashMap<>();
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
@Override
public <T> boolean addOutputPayloadVerifier(String topic, Predicate<T> verifier) {
Predicate<Message<?>> messagePredicate = message -> verifier.test((T) message.getPayload());
return addOutputMessageVerifier(topic, messagePredicate);
}
@Override
public boolean addOutputMessageVerifier(String topic, Predicate<Message<?>> 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<Message<?>> 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<Verifier> l) -> l.forEach(v -> v.setSatisfied(false)));
}
protected abstract Function<Message<?>, 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<Message<?>> {
private final Predicate<Message<?>> predicate;
private final AtomicBoolean satisfied;
private Verifier(Predicate<Message<?>> 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();
}
}
}

101
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java
View File

@@ -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<OutputFrame> {
private static Logger logger = LoggerFactory.getLogger(LogMatcher.class);
private List<Consumer<String>> listeners = new LinkedList<>();
protected AtomicBoolean matched = new AtomicBoolean();
public Callable<Boolean> verifies(Consumer<LogListener> 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<Boolean> 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<String> {
private AtomicBoolean matched = new AtomicBoolean();
public final static class CountingLogMatcher extends LogMatcher {
private Optional<Runnable> 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<Boolean> matches() {
return () -> count.get() == 0;
}
}
}

116
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainer.java Normal file
View File

@@ -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<StreamAppContainer> {
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();
}

32
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamIApplicationIntegrationTestSupport.java → applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApplicationIntegrationTestSupport.java
View File

@@ -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<Boolean> verifyOutputMessages() {
return () -> testListener.isVerified().get();
}
protected <P> Callable<Boolean> verifyOutputPayload(Predicate<P> outputVerifier) {
testListener.addOutputPayloadVerifier(outputVerifier);
return () -> testListener.isVerified().get();
}
protected Callable<Boolean> verifyOutputMessage(Predicate<Message<?>> outputVerifier) {
testListener.addOutputMessageVerifier(outputVerifier);
return () -> testListener.isVerified().get();
}
}

50
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamApps.java
View File

@@ -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<String>) 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<String>) env -> logger.debug(env));
container.withLogConsumer(appLog(container.getImage().get()));
});
}
logger.debug("\nSink container environment:");
sinkContainer().getEnv().forEach((Consumer<String>) 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<String>) 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<String>) env -> logger.debug(env));
});
}
logger.debug("\nSink container environment for {} :", sinkContainer().getImage().get());
sinkContainer().getEnv().forEach((Consumer<String>) env -> logger.debug(env));
}
public static abstract class Builder<S extends StreamApps> {
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<String, String> binderProperties();
protected abstract S doBuild(GenericContainer sourceContainer,
List<GenericContainer> 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<String, String> binderProperties();
}
}

104
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/TestTopicListener.java Normal file
View File

@@ -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 <P> the expected payload type
* @return true if it is registered, false if it is already registered.
*/
<P> boolean addOutputPayloadVerifier(String topic, Predicate<P> outputVerifier);
/**
* Register a Message payload verifier on the default destination.
* @param outputVerifier a {code Predicate} to test the payload.
* @param <P> the expected payload type
* @return true if it is registered, false if it is already registered.
*/
default <P> boolean addOutputPayloadVerifier(Predicate<P> 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<Message<?>> 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<Message<?>> 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);
}

34
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/AbstractKafkaStreamApplicationIntegrationTests.java → applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainer.java
View File

@@ -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;
}
}

205
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupport.java Normal file
View File

@@ -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<String, String> kafkaTemplate(ProducerFactory producerFactory) {
return new KafkaTemplate(producerFactory);
}
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> 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<String, String> cf = new DefaultKafkaConsumerFactory<>(configs);
cf.setBootstrapServersSupplier(() -> kafka.getBootstrapServers());
return cf;
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
ConsumerFactory consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
return factory;
}
@Bean
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> configs = new HashMap<>();
configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(configs);
pf.setBootstrapServersSupplier(() -> kafka.getBootstrapServers());
return pf;
}
@Bean
public AdminClient admin() {
Map<String, Object> 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<String> topics = admin.listTopics().names().get();
return topics.contains(STREAM_APPLICATIONS_TEST_TOPIC);
});
}
@Override
public boolean addOutputMessageVerifier(String topic, Predicate<Message<?>> 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<Message<?>, String> topicForMessage() {
return message -> (String) message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC);
}
@KafkaHandler(isDefault = true)
public void listen(Message<?> message) {
super.listen(message);
}
}
}
}

6
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApps.java
View File

@@ -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> kafkaStreamApps(String streamName, GenericContainer messageBrokerContainer) {
return new KafkaBuilder(streamName, messageBrokerContainer);
}
public static final class KafkaBuilder extends Builder {
public static final class KafkaBuilder extends Builder<KafkaStreamApps> {
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<GenericContainer> processorContainers, GenericContainer sinkContainer) {
return new KafkaStreamApps(sourceContainer, processorContainers, sinkContainer);
}

44
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainer.java Normal file
View File

@@ -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;
}
}

218
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupport.java Normal file
View File

@@ -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<String, Set<Message<?>>> 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<Message<?>> messages = cache.getIfPresent(topic);
if (messages.add(message)) {
logger.debug("Caching message: {} for topic {}", message, topic);
}
}
@Override
protected Function<Message<?>, 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);
}
}
}
}
}

60
applications/stream-applications-core/common/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApps.java Normal file
View File

@@ -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<GenericContainer> processorContainers,
GenericContainer sinkContainer) {
super(sourceContainer, processorContainers, sinkContainer);
}
public static Builder<RabbitMQStreamApps> rabbitMQStreamApps(String streamName,
GenericContainer messageBrokerContainer) {
return new RabbitMQBuilder(streamName, messageBrokerContainer);
}
public static final class RabbitMQBuilder extends Builder<RabbitMQStreamApps> {
protected RabbitMQBuilder(String streamName, GenericContainer messageBrokerContainer) {
super(streamName, messageBrokerContainer);
}
protected Map<String, String> binderProperties() {
return fluentMap()
.withEntry("SPRING_RABBITMQ_HOST",
messageBrokerContainer.getNetworkAliases().get(0))
.withEntry("SPRING_RABBITMQ_PORT", "5672");
}
@Override
protected RabbitMQStreamApps doBuild(GenericContainer sourceContainer,
List<GenericContainer> processorContainers, GenericContainer sinkContainer) {
return new RabbitMQStreamApps(sourceContainer, processorContainers, sinkContainer);
}
}
}

74
applications/stream-applications-core/common/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamApplicationIntegrationTestSupportTests.java Normal file
View File

@@ -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"))));
}
}

74
applications/stream-applications-core/common/stream-applications-test-support/src/test/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamApplicationIntegrationTestSupportTests.java Normal file
View File

@@ -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"))));
}
}

17
applications/stream-applications-core/common/stream-applications-test-support/src/test/resources/logback-test.xml Normal file
View File

@@ -0,0 +1,17 @@
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
</encoder>
</appender>
<root level="info">
<appender-ref ref="STDOUT"/>
</root>
<logger name="org.testcontainers" level="INFO"/>
<logger name="com.github.dockerjava" level="WARN"/>
<!-- <logger name="org.springframework.amqp.rabbit.core" level="DEBUG"/>-->
<!-- <logger name="org.springframework.kafka" level="INFO"/>-->
<logger name="org.springframework.cloud.stream.app.test.integration" level="DEBUG"/>
</configuration>

635
stream-applications-build/pom.xml
View File

@@ -1,351 +1,356 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>stream-applications-build</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>stream-applications-build</name>
<description>Common Parent for Functions and Applications</description>
<packaging>pom</packaging>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>stream-applications-build</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>stream-applications-build</name>
<description>Common Parent for Functions and Applications</description>
<packaging>pom</packaging>
<properties>
<java.version>1.8</java.version>
<maven-javadoc-plugin.version>3.1.1</maven-javadoc-plugin.version>
<maven-source-plugin.version>3.2.1</maven-source-plugin.version>
<maven-surefire-plugin.version>2.22.2</maven-surefire-plugin.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<maven.compiler.source>${java.version}</maven.compiler.source>
<maven.compiler.target>${java.version}</maven.compiler.target>
<maven-source-plugin.version>3.0.1</maven-source-plugin.version>
<maven-checkstyle-plugin.version>3.1.0</maven-checkstyle-plugin.version>
<disable.checks>false</disable.checks>
<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>
<maven-checkstyle-plugin.failOnViolation>true</maven-checkstyle-plugin.failOnViolation>
<maven-checkstyle-plugin.includeTestSourceDirectory>true</maven-checkstyle-plugin.includeTestSourceDirectory>
<puppycrawl-tools-checkstyle.version>8.29</puppycrawl-tools-checkstyle.version>
<checkstyle.location>https://raw.githubusercontent.com/spring-cloud/stream-applications/master/etc/checkstyle</checkstyle.location>
<checkstyle.suppressions.file>
${checkstyle.location}/checkstyle-suppressions.xml
</checkstyle.suppressions.file>
<checkstyle.nohttp.file>
${checkstyle.location}/nohttp-checkstyle.xml
</checkstyle.nohttp.file>
<checkstyle.additional.suppressions.file>
${checkstyle.location}/checkstyle-suppressions.xml
</checkstyle.additional.suppressions.file>
<nohttp-checkstyle.version>0.0.2.RELEASE</nohttp-checkstyle.version>
<disable.nohttp.checks>true</disable.nohttp.checks>
<spring-javaformat-checkstyle.version>0.0.7</spring-javaformat-checkstyle.version>
<spring-boot.version>2.3.4.RELEASE</spring-boot.version>
<spring-cloud-function.version>3.0.9.RELEASE</spring-cloud-function.version>
<spring-integration-dependencies.version>5.3.3.BUILD-SNAPSHOT</spring-integration-dependencies.version>
<test-containers.version>1.15.0-rc2</test-containers.version>
<maven-flatten-plugin.version>1.2.5</maven-flatten-plugin.version>
<maven-surefire-plugin.version>2.22.2</maven-surefire-plugin.version>
</properties>
<properties>
<java.version>1.8</java.version>
<maven-javadoc-plugin.version>3.1.1</maven-javadoc-plugin.version>
<maven-source-plugin.version>3.2.1</maven-source-plugin.version>
<maven-surefire-plugin.version>2.22.2</maven-surefire-plugin.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<maven.compiler.source>${java.version}</maven.compiler.source>
<maven.compiler.target>${java.version}</maven.compiler.target>
<maven-source-plugin.version>3.0.1</maven-source-plugin.version>
<maven-checkstyle-plugin.version>3.1.0</maven-checkstyle-plugin.version>
<disable.checks>false</disable.checks>
<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>
<maven-checkstyle-plugin.failOnViolation>true</maven-checkstyle-plugin.failOnViolation>
<maven-checkstyle-plugin.includeTestSourceDirectory>true</maven-checkstyle-plugin.includeTestSourceDirectory>
<puppycrawl-tools-checkstyle.version>8.29</puppycrawl-tools-checkstyle.version>
<checkstyle.location>https://raw.githubusercontent.com/spring-cloud/stream-applications/master/etc/checkstyle
</checkstyle.location>
<checkstyle.suppressions.file>
${checkstyle.location}/checkstyle-suppressions.xml
</checkstyle.suppressions.file>
<checkstyle.nohttp.file>
${checkstyle.location}/nohttp-checkstyle.xml
</checkstyle.nohttp.file>
<checkstyle.additional.suppressions.file>
${checkstyle.location}/checkstyle-suppressions.xml
</checkstyle.additional.suppressions.file>
<nohttp-checkstyle.version>0.0.2.RELEASE</nohttp-checkstyle.version>
<disable.nohttp.checks>true</disable.nohttp.checks>
<spring-javaformat-checkstyle.version>0.0.7</spring-javaformat-checkstyle.version>
<spring-boot.version>2.3.4.RELEASE</spring-boot.version>
<spring-kafka.version>2.6.0-M1</spring-kafka.version>
<spring-rabbit.version>2.2.11.RELEASE</spring-rabbit.version>
<spring-cloud-function.version>3.0.9.RELEASE</spring-cloud-function.version>
<spring-integration-dependencies.version>5.3.3.BUILD-SNAPSHOT</spring-integration-dependencies.version>
<test-containers.version>1.15.0-rc2</test-containers.version>
<maven-flatten-plugin.version>1.2.5</maven-flatten-plugin.version>
<maven-surefire-plugin.version>2.22.2</maven-surefire-plugin.version>
</properties>
<dependencyManagement>
<dependencies>
<!-- Adding SI bom here is temporary to address some issues. It should come from Boot dependencies -->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-bom</artifactId>
<version>${spring-integration-dependencies.version}</version>
<scope>import</scope>
<type>pom</type>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-function-dependencies</artifactId>
<version>${spring-cloud-function.version}</version>
<scope>import</scope>
<type>pom</type>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>${spring-boot.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencyManagement>
<dependencies>
<!-- Adding SI bom here is temporary to address some issues. It should come from Boot dependencies -->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-bom</artifactId>
<version>${spring-integration-dependencies.version}</version>
<scope>import</scope>
<type>pom</type>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-function-dependencies</artifactId>
<version>${spring-cloud-function.version}</version>
<scope>import</scope>
<type>pom</type>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>${spring-boot.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>flatten-maven-plugin</artifactId>
<version>${maven-flatten-plugin.version}</version>
<configuration>
<updatePomFile>true</updatePomFile>
<flattenMode>resolveCiFriendliesOnly</flattenMode>
</configuration>
<executions>
<execution>
<id>flatten</id>
<phase>process-resources</phase>
<goals>
<goal>flatten</goal>
</goals>
</execution>
<execution>
<id>flatten.clean</id>
<phase>clean</phase>
<goals>
<goal>clean</goal>
</goals>
</execution>
</executions>
</plugin>
<build>
<plugins>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>flatten-maven-plugin</artifactId>
<version>${maven-flatten-plugin.version}</version>
<configuration>
<updatePomFile>true</updatePomFile>
<flattenMode>resolveCiFriendliesOnly</flattenMode>
</configuration>
<executions>
<execution>
<id>flatten</id>
<phase>process-resources</phase>
<goals>
<goal>flatten</goal>
</goals>
</execution>
<execution>
<id>flatten.clean</id>
<phase>clean</phase>
<goals>
<goal>clean</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-javadoc-plugin</artifactId>
<version>${maven-javadoc-plugin.version}</version>
<executions>
<execution>
<id>javadoc</id>
<phase>package</phase>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
<configuration>
<quiet>true</quiet>
</configuration>
</plugin>
<plugin>
<artifactId>maven-javadoc-plugin</artifactId>
<version>${maven-javadoc-plugin.version}</version>
<executions>
<execution>
<id>javadoc</id>
<phase>package</phase>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
<configuration>
<quiet>true</quiet>
</configuration>
</plugin>
<plugin>
<artifactId>maven-source-plugin</artifactId>
<version>${maven-source-plugin.version}</version>
<executions>
<execution>
<id>attach-sources</id>
<phase>package</phase>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-source-plugin</artifactId>
<version>${maven-source-plugin.version}</version>
<executions>
<execution>
<id>attach-sources</id>
<phase>package</phase>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>${maven-surefire-plugin.version}</version>
<configuration>
<includes>
<include>**/*Tests.java</include>
<include>**/*Test.java</include>
</includes>
<excludes>
<exclude>**/Abstract*.java</exclude>
</excludes>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>${maven-surefire-plugin.version}</version>
<configuration>
<includes>
<include>**/*Tests.java</include>
<include>**/*Test.java</include>
</includes>
<excludes>
<exclude>**/Abstract*.java</exclude>
</excludes>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<version>${maven-checkstyle-plugin.version}</version>
<dependencies>
<dependency>
<groupId>com.puppycrawl.tools</groupId>
<artifactId>checkstyle</artifactId>
<version>${puppycrawl-tools-checkstyle.version}</version>
</dependency>
<dependency>
<groupId>io.spring.javaformat</groupId>
<artifactId>spring-javaformat-checkstyle</artifactId>
<version>${spring-javaformat-checkstyle.version}</version>
</dependency>
<dependency>
<groupId>io.spring.nohttp</groupId>
<artifactId>nohttp-checkstyle</artifactId>
<version>${nohttp-checkstyle.version}</version>
</dependency>
</dependencies>
<executions>
<execution>
<id>checkstyle-validation</id>
<phase>validate</phase>
<inherited>true</inherited>
<configuration>
<skip>${disable.checks}</skip>
<configLocation>${checkstyle.location}/checkstyle.xml</configLocation>
<headerLocation>${checkstyle.location}/checkstyle-header.txt</headerLocation>
<propertyExpansion>
checkstyle.build.directory=${project.build.directory}
checkstyle.suppressions.file=${checkstyle.suppressions.file}
checkstyle.additional.suppressions.file=${checkstyle.additional.suppressions.file}
</propertyExpansion>
<consoleOutput>true</consoleOutput>
<includeTestSourceDirectory>
${maven-checkstyle-plugin.includeTestSourceDirectory}
</includeTestSourceDirectory>
<failsOnError>${maven-checkstyle-plugin.failsOnError}
</failsOnError>
<failOnViolation>
${maven-checkstyle-plugin.failOnViolation}
</failOnViolation>
</configuration>
<goals>
<goal>check</goal>
</goals>
</execution>
<execution>
<id>no-http-checkstyle-validation</id>
<phase>validate</phase>
<inherited>true</inherited>
<configuration>
<skip>${disable.nohttp.checks}</skip>
<configLocation>${checkstyle.nohttp.file}</configLocation>
<includes>**/*</includes>
<excludes>**/.idea/**/*,**/.git/**/*,**/target/**/*,**/*.log</excludes>
<sourceDirectories>./</sourceDirectories>
</configuration>
<goals>
<goal>check</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<version>${maven-checkstyle-plugin.version}</version>
<dependencies>
<dependency>
<groupId>com.puppycrawl.tools</groupId>
<artifactId>checkstyle</artifactId>
<version>${puppycrawl-tools-checkstyle.version}</version>
</dependency>
<dependency>
<groupId>io.spring.javaformat</groupId>
<artifactId>spring-javaformat-checkstyle</artifactId>
<version>${spring-javaformat-checkstyle.version}</version>
</dependency>
<dependency>
<groupId>io.spring.nohttp</groupId>
<artifactId>nohttp-checkstyle</artifactId>
<version>${nohttp-checkstyle.version}</version>
</dependency>
</dependencies>
<executions>
<execution>
<id>checkstyle-validation</id>
<phase>validate</phase>
<inherited>true</inherited>
<configuration>
<skip>${disable.checks}</skip>
<configLocation>${checkstyle.location}/checkstyle.xml</configLocation>
<headerLocation>${checkstyle.location}/checkstyle-header.txt</headerLocation>
<propertyExpansion>
checkstyle.build.directory=${project.build.directory}
checkstyle.suppressions.file=${checkstyle.suppressions.file}
checkstyle.additional.suppressions.file=${checkstyle.additional.suppressions.file}
</propertyExpansion>
<consoleOutput>true</consoleOutput>
<includeTestSourceDirectory>
${maven-checkstyle-plugin.includeTestSourceDirectory}
</includeTestSourceDirectory>
<failsOnError>${maven-checkstyle-plugin.failsOnError}
</failsOnError>
<failOnViolation>
${maven-checkstyle-plugin.failOnViolation}
</failOnViolation>
</configuration>
<goals>
<goal>check</goal>
</goals>
</execution>
<execution>
<id>no-http-checkstyle-validation</id>
<phase>validate</phase>
<inherited>true</inherited>
<configuration>
<skip>${disable.nohttp.checks}</skip>
<configLocation>${checkstyle.nohttp.file}</configLocation>
<includes>**/*</includes>
<excludes>**/.idea/**/*,**/.git/**/*,**/target/**/*,**/*.log</excludes>
<sourceDirectories>./</sourceDirectories>
</configuration>
<goals>
<goal>check</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</plugins>
</build>
</build>
<licenses>
<licenses>
<license>
<name>Apache License, Version 2.0</name>
<url>http://www.apache.org/licenses/LICENSE-2.0</url>
<comments>Copyright 2014-2020 the original author or authors.
<license>
<name>Apache License, Version 2.0</name>
<url>http://www.apache.org/licenses/LICENSE-2.0</url>
<comments>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.</comments>
</license>
See the License for the specific language governing permissions and
limitations under the License.
</comments>
</license>
</licenses>
</licenses>
<scm>
<connection>scm:git:git://github.com/pivotal/java-functions.git</connection>
<developerConnection>scm:git:ssh://git@github.com/pivotal/java-functions.git</developerConnection>
<url>https://github.com/pivotal/java-functions</url>
</scm>
<scm>
<connection>scm:git:git://github.com/pivotal/java-functions.git</connection>
<developerConnection>scm:git:ssh://git@github.com/pivotal/java-functions.git</developerConnection>
<url>https://github.com/pivotal/java-functions</url>
</scm>
<distributionManagement>
<distributionManagement>
<repository>
<id>repo.spring.io</id>
<name>Spring Release Repository</name>
<url>https://repo.spring.io/libs-release-local</url>
</repository>
<repository>
<id>repo.spring.io</id>
<name>Spring Release Repository</name>
<url>https://repo.spring.io/libs-release-local</url>
</repository>
<snapshotRepository>
<id>repo.spring.io</id>
<name>Spring Snapshot Repository</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</snapshotRepository>
<snapshotRepository>
<id>repo.spring.io</id>
<name>Spring Snapshot Repository</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</snapshotRepository>
</distributionManagement>
</distributionManagement>
<repositories>
<repositories>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
</repository>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
</repository>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
</repository>
</repositories>
</repositories>
<pluginRepositories>
<pluginRepositories>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
</pluginRepository>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
</pluginRepository>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
</pluginRepository>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
</pluginRepositories>
<profiles>
<profiles>
<profile>
<id>milestone</id>
<distributionManagement>
<repository>
<id>repo.spring.io</id>
<name>Spring Milestone Repository</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</repository>
</distributionManagement>
</profile>
<profile>
<id>milestone</id>
<distributionManagement>
<repository>
<id>repo.spring.io</id>
<name>Spring Milestone Repository</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</repository>
</distributionManagement>
</profile>
<profile>
<id>central</id>
<build>
<plugins>
<plugin>
<artifactId>maven-gpg-plugin</artifactId>
<executions>
<execution>
<id>sign-artifacts</id>
<phase>verify</phase>
<goals>
<goal>sign</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
<distributionManagement>
<repository>
<id>sonatype-nexus-staging</id>
<name>Nexus Release Repository</name>
<url>https://oss.sonatype.org/service/local/staging/deploy/maven2/</url>
</repository>
<snapshotRepository>
<id>sonatype-nexus-snapshots</id>
<name>Sonatype Nexus Snapshots</name>
<url>https://oss.sonatype.org/content/repositories/snapshots/</url>
</snapshotRepository>
</distributionManagement>
</profile>
<profile>
<id>central</id>
<build>
<plugins>
<plugin>
<artifactId>maven-gpg-plugin</artifactId>
<executions>
<execution>
<id>sign-artifacts</id>
<phase>verify</phase>
<goals>
<goal>sign</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
<distributionManagement>
<repository>
<id>sonatype-nexus-staging</id>
<name>Nexus Release Repository</name>
<url>https://oss.sonatype.org/service/local/staging/deploy/maven2/</url>
</repository>
<snapshotRepository>
<id>sonatype-nexus-snapshots</id>
<name>Sonatype Nexus Snapshots</name>
<url>https://oss.sonatype.org/content/repositories/snapshots/</url>
</snapshotRepository>
</distributionManagement>
</profile>
</profiles>
</profiles>
</project>