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

Remove schema and reactive modules

Browse Source
This commit is contained in:
Oleg Zhurakousky
2020-02-14 10:36:20 +01:00
parent 54b9fb63a7
commit 6d2a07c601
30 changed files with 0 additions and 3308 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
spring-cloud-stream-reactive/.jdk8
View File

50
spring-cloud-stream-reactive/pom.xml
View File

@@ -1,50 +0,0 @@
<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-parent</artifactId>
<version>3.0.3.BUILD-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-reactive</artifactId>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure-processor</artifactId>
<optional>true</optional>
</dependency>
</dependencies>
</project>

68
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/DefaultFluxSender.java
View File

@@ -1,68 +0,0 @@
/*
* Copyright 2017-2019 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.reactive;
import java.util.function.Consumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import org.springframework.util.Assert;
/**
* Default {@link org.springframework.cloud.stream.reactive.FluxSender} implementation.
* This implementation may be used for cancelling a subscription on the underlying
* {@link reactor.core.publisher.Flux}.
*
* @author Soby Chacko
* @since 1.3.0
*/
class DefaultFluxSender implements FluxSender {
private final Consumer<Object> consumer;
private Log log = LogFactory.getLog(DefaultFluxSender.class);
private volatile Disposable disposable;
DefaultFluxSender(Consumer<Object> consumer) {
Assert.notNull(consumer, "Consumer must not be null");
this.consumer = consumer;
}
@Override
public Mono<Void> send(Flux<?> flux) {
MonoProcessor<Void> sendResult = MonoProcessor.create();
// add error handling and reconnect in the event of an error
this.disposable = flux
.doOnError(e -> this.log.error("Error during processing: ", e)).retry()
.subscribe(this.consumer, sendResult::onError, sendResult::onComplete);
return sendResult;
}
@Override
public void close() {
if (this.disposable != null) {
this.disposable.dispose();
}
}
}

44
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/FluxSender.java
View File

@@ -1,44 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.io.Closeable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Used for {@link org.springframework.cloud.stream.annotation.StreamListener} and
* {@link org.springframework.cloud.stream.reactive.StreamEmitter} arguments annotated
* with {@link org.springframework.cloud.stream.annotation.Output}.
*
* @author Marius Bogoevici
* @see reactor.core.Disposable
*/
public interface FluxSender extends Closeable {
/**
* Streams the {@link reactor.core.publisher.Flux} through the binding target
* corresponding to the {@link org.springframework.cloud.stream.annotation.Output}
* annotation of the argument.
* @param flux a {@link Flux} that will be streamed through the binding target
* @return a {@link Mono} representing the result of sending the flux (completion or
* error)
*/
Mono<Void> send(Flux<?> flux);
}

50
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToFluxSenderParameterAdapter.java
View File

@@ -1,50 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
import org.springframework.core.MethodParameter;
import org.springframework.core.ResolvableType;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
/**
* Adapts an {@link org.springframework.cloud.stream.annotation.Output} annotated
* {@link FluxSender} to an outbound {@link MessageChannel}.
*
* @author Marius Bogoevici
* @author Soby Chacko
*/
public class MessageChannelToFluxSenderParameterAdapter
implements StreamListenerParameterAdapter<FluxSender, MessageChannel> {
@Override
public boolean supports(Class<?> bindingTargetType, MethodParameter methodParameter) {
ResolvableType type = ResolvableType.forMethodParameter(methodParameter);
return MessageChannel.class.isAssignableFrom(bindingTargetType)
&& FluxSender.class.isAssignableFrom(type.getRawClass());
}
@Override
public FluxSender adapt(MessageChannel bindingTarget, MethodParameter parameter) {
return new DefaultFluxSender(result -> bindingTarget
.send(result instanceof Message<?> ? (Message<?>) result
: MessageBuilder.withPayload(result).build()));
}
}

121
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java
View File

@@ -1,121 +0,0 @@
/*
* Copyright 2016-2019 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.reactive;
import reactor.core.publisher.Flux;
import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
import org.springframework.core.MethodParameter;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.ResolvableType;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
/**
* Adapts an {@link org.springframework.cloud.stream.annotation.Input} annotated
* {@link MessageChannel} to a {@link Flux}.
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Vinicius Carvalho
*/
public class MessageChannelToInputFluxParameterAdapter
implements StreamListenerParameterAdapter<Flux<?>, SubscribableChannel> {
private final CompositeMessageConverter messageConverter;
public MessageChannelToInputFluxParameterAdapter(
CompositeMessageConverter messageConverter) {
Assert.notNull(messageConverter, "cannot not be null");
this.messageConverter = messageConverter;
}
@Override
public boolean supports(Class<?> bindingTargetType, MethodParameter methodParameter) {
return MessageChannel.class.isAssignableFrom(bindingTargetType)
&& Flux.class.isAssignableFrom(methodParameter.getParameterType());
}
@Override
public Flux<?> adapt(final SubscribableChannel bindingTarget,
MethodParameter parameter) {
final ResolvableType fluxResolvableType = ResolvableType
.forMethodParameter(parameter);
final ResolvableType fluxTypeParameter = fluxResolvableType.getGeneric(0);
final Class<?> fluxTypeParameterRawClass = fluxTypeParameter.getRawClass();
final Class<?> fluxTypeParameterClass = (fluxTypeParameterRawClass != null)
? fluxTypeParameterRawClass : Object.class;
final Object monitor = new Object();
Object conversionHint = ParameterizedTypeReference
.forType(fluxTypeParameter.getType());
if (Message.class.isAssignableFrom(fluxTypeParameterClass)) {
final ResolvableType payloadTypeParameter = fluxTypeParameter.getGeneric(0);
final Class<?> payloadTypeParameterRawClass = payloadTypeParameter
.getRawClass();
final Class<?> payloadTypeParameterClass = (payloadTypeParameterRawClass != null)
? payloadTypeParameterRawClass : Object.class;
return Flux.create(emitter -> {
MessageHandler messageHandler = message -> {
synchronized (monitor) {
if (payloadTypeParameterClass
.isAssignableFrom(message.getPayload().getClass())) {
emitter.next(message);
}
else {
emitter.next(MessageBuilder.createMessage(
this.messageConverter.fromMessage(message,
payloadTypeParameterClass, conversionHint),
message.getHeaders()));
}
}
};
bindingTarget.subscribe(messageHandler);
emitter.onCancel(() -> bindingTarget.unsubscribe(messageHandler));
}).publish().autoConnect();
}
else {
return Flux.create(emitter -> {
MessageHandler messageHandler = message -> {
synchronized (monitor) {
if (fluxTypeParameterClass
.isAssignableFrom(message.getPayload().getClass())) {
emitter.next(message.getPayload());
}
else {
emitter.next(this.messageConverter.fromMessage(message,
fluxTypeParameterClass, conversionHint));
}
}
};
bindingTarget.subscribe(messageHandler);
emitter.onCancel(() -> bindingTarget.unsubscribe(messageHandler));
}).publish().autoConnect();
}
}
}

63
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/PublisherToMessageChannelResultAdapter.java
View File

@@ -1,63 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.io.Closeable;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import org.springframework.cloud.stream.binding.StreamListenerResultAdapter;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
/**
* A {@link org.springframework.cloud.stream.binding.StreamListenerResultAdapter} from a
* {@link Publisher} return type to a bound {@link MessageChannel}.
*
* @author Marius Bogoevici
* @author Soby Chacko
*
*/
public class PublisherToMessageChannelResultAdapter
implements StreamListenerResultAdapter<Publisher<?>, MessageChannel> {
private Log log = LogFactory.getLog(PublisherToMessageChannelResultAdapter.class);
@Override
public boolean supports(Class<?> resultType, Class<?> bindingTarget) {
return Publisher.class.isAssignableFrom(resultType)
&& MessageChannel.class.isAssignableFrom(bindingTarget);
}
public Closeable adapt(Publisher<?> streamListenerResult,
MessageChannel bindingTarget) {
Disposable disposable = Flux.from(streamListenerResult)
.doOnError(e -> this.log.error("Error while processing result", e))
.retry()
.subscribe(result -> bindingTarget
.send(result instanceof Message<?> ? (Message<?>) result
: MessageBuilder.withPayload(result).build()));
return disposable::dispose;
}
}

58
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java
View File

@@ -1,58 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.cloud.stream.binding.BindingService;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author Marius Bogoevici
*/
@Configuration
@ConditionalOnBean(BindingService.class)
public class ReactiveSupportAutoConfiguration {
@Bean
public static StreamEmitterAnnotationBeanPostProcessor streamEmitterAnnotationBeanPostProcessor() {
return new StreamEmitterAnnotationBeanPostProcessor();
}
@Bean
@ConditionalOnMissingBean(MessageChannelToInputFluxParameterAdapter.class)
public MessageChannelToInputFluxParameterAdapter messageChannelToInputFluxArgumentAdapter(
CompositeMessageConverterFactory compositeMessageConverterFactory) {
return new MessageChannelToInputFluxParameterAdapter(
compositeMessageConverterFactory.getMessageConverterForAllRegistered());
}
@Bean
@ConditionalOnMissingBean(MessageChannelToFluxSenderParameterAdapter.class)
public MessageChannelToFluxSenderParameterAdapter messageChannelToFluxSenderArgumentAdapter() {
return new MessageChannelToFluxSenderParameterAdapter();
}
@Bean
@ConditionalOnMissingBean(PublisherToMessageChannelResultAdapter.class)
public PublisherToMessageChannelResultAdapter fluxToMessageChannelResultAdapter() {
return new PublisherToMessageChannelResultAdapter();
}
}

97
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitter.java
View File

@@ -1,97 +0,0 @@
/*
* Copyright 2017-2019 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.reactive;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
/**
* Method level annotation that marks a method to be an emitter to outputs declared via
* {@link EnableBinding} (e.g. channels).
*
* This annotation is intended to be used in a Spring Cloud Stream application that
* requires a source to write to one or more {@link Output}s using the reactive paradigm.
*
* No {@link Input}s are allowed on a method that is annotated with StreamEmitter.
*
* Depending on how the method is structured, there are some flexibility in how the
* {@link Output} may be used.
*
* Here are some supported usage patterns:
*
* A StreamEmitter method that has a return type cannot take any method parameters.
*
* <pre class="code">
* &#064;StreamEmitter
* &#064;Output(Source.OUTPUT)
* public Flux&lt;String&gt; emit() {
* return Flux.intervalMillis(1000)
* .map(l -> "Hello World!!");
* }
* </pre>
*
* The following examples show how a void return type can be used on a method with
* StreamEmitter and how the method signatures could be used in a flexible manner.
*
* <pre class="code">
* &#064;StreamEmitter
* public void emit(&#064;Output(Source.OUTPUT) FluxSender output) {
* output.send(Flux.intervalMillis(1000)
* .map(l -> "Hello World!!"));
* }
* </pre>
*
* <pre class="code">
* &#064;StreamEmitter
* &#064;Output(Source.OUTPUT)
* public void emit(FluxSender output) {
* output.send(Flux.intervalMillis(1000)
* .map(l -> "Hello World!!"));
* }
* </pre>
*
* <pre class="code">
* &#064;StreamEmitter
* public void emit(&#064;Output("OUTPUT1") FluxSender output1,
* &#064;Output("OUTPUT2") FluxSender output2,
* &#064;Output("OUTPUT3)" FluxSender output3) {
* output1.send(Flux.intervalMillis(1000)
* .map(l -> "Hello World!!"));
* output2.send(Flux.intervalMillis(1000)
* .map(l -> "Hello World!!"));
* output3.send(Flux.intervalMillis(1000)
* .map(l -> "Hello World!!"));
* }
*</pre>
*
* @author Soby Chacko
*
* @since 1.3.0
*/
@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface StreamEmitter {
}

294
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitterAnnotationBeanPostProcessor.java
View File

@@ -1,294 +0,0 @@
/*
* Copyright 2017-2019 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.reactive;
import java.io.Closeable;
import java.io.IOException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.binding.MessageChannelStreamListenerResultAdapter;
import org.springframework.cloud.stream.binding.StreamAnnotationCommonMethodUtils;
import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
import org.springframework.cloud.stream.binding.StreamListenerResultAdapter;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.MethodParameter;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.annotation.SynthesizingMethodParameter;
import org.springframework.util.Assert;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* {@link BeanPostProcessor} that handles {@link StreamEmitter} annotations found on bean
* methods.
*
* @author Soby Chacko
* @author Artem Bilan
* @since 1.3.0
*/
public class StreamEmitterAnnotationBeanPostProcessor implements BeanPostProcessor,
SmartInitializingSingleton, ApplicationContextAware, SmartLifecycle {
private static final Log log = LogFactory
.getLog(StreamEmitterAnnotationBeanPostProcessor.class);
private final List<Closeable> closeableFluxResources = new ArrayList<>();
private final Lock lock = new ReentrantLock();
@SuppressWarnings("rawtypes")
private Collection<StreamListenerParameterAdapter> parameterAdapters;
@SuppressWarnings("rawtypes")
private Collection<StreamListenerResultAdapter> resultAdapters;
private ConfigurableApplicationContext applicationContext;
private MultiValueMap<Object, Method> mappedStreamEmitterMethods = new LinkedMultiValueMap<>();
private volatile boolean running;
private static void validateStreamEmitterMethod(Method method,
int outputAnnotationCount, String methodAnnotatedOutboundName) {
if (StringUtils.hasText(methodAnnotatedOutboundName)) {
Assert.isTrue(outputAnnotationCount == 0,
StreamEmitterErrorMessages.INVALID_OUTPUT_METHOD_PARAMETERS);
}
else {
Assert.isTrue(outputAnnotationCount > 0,
StreamEmitterErrorMessages.NO_OUTPUT_SPECIFIED);
}
if (!method.getReturnType().equals(Void.TYPE)) {
Assert.isTrue(StringUtils.hasText(methodAnnotatedOutboundName),
StreamEmitterErrorMessages.RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
Assert.isTrue(method.getParameterCount() == 0,
StreamEmitterErrorMessages.RETURN_TYPE_METHOD_ARGUMENTS);
}
else {
if (!StringUtils.hasText(methodAnnotatedOutboundName)) {
int methodArgumentsLength = method.getParameterTypes().length;
for (int parameterIndex = 0; parameterIndex < methodArgumentsLength; parameterIndex++) {
MethodParameter methodParameter = new MethodParameter(method,
parameterIndex);
if (methodParameter.hasParameterAnnotation(Output.class)) {
String outboundName = (String) AnnotationUtils.getValue(
methodParameter.getParameterAnnotation(Output.class));
Assert.isTrue(StringUtils.hasText(outboundName),
StreamEmitterErrorMessages.INVALID_OUTBOUND_NAME);
}
else {
throw new IllegalArgumentException(
StreamEmitterErrorMessages.OUTPUT_ANNOTATION_MISSING_ON_METHOD_PARAMETERS_VOID_RETURN_TYPE);
}
}
}
}
}
@Override
public final void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
Assert.isTrue(applicationContext instanceof ConfigurableApplicationContext,
"ConfigurableApplicationContext is required");
this.applicationContext = (ConfigurableApplicationContext) applicationContext;
}
@Override
public void afterSingletonsInstantiated() {
this.parameterAdapters = this.applicationContext
.getBeansOfType(StreamListenerParameterAdapter.class).values();
this.resultAdapters = new ArrayList<>(this.applicationContext
.getBeansOfType(StreamListenerResultAdapter.class).values());
this.resultAdapters.add(new MessageChannelStreamListenerResultAdapter());
}
@Override
public Object postProcessAfterInitialization(final Object bean, final String beanName)
throws BeansException {
Class<?> targetClass = AopUtils.getTargetClass(bean);
ReflectionUtils.doWithMethods(targetClass, method -> {
if (AnnotatedElementUtils.isAnnotated(method, StreamEmitter.class)) {
this.mappedStreamEmitterMethods.add(bean, method);
}
}, ReflectionUtils.USER_DECLARED_METHODS);
return bean;
}
@Override
public void start() {
try {
this.lock.lock();
if (!this.running) {
this.mappedStreamEmitterMethods.forEach((k, v) -> v.forEach(item -> {
Assert.isTrue(item.getAnnotation(Input.class) == null,
StreamEmitterErrorMessages.INPUT_ANNOTATIONS_ARE_NOT_ALLOWED);
String methodAnnotatedOutboundName = StreamAnnotationCommonMethodUtils
.getOutboundBindingTargetName(item);
int outputAnnotationCount = StreamAnnotationCommonMethodUtils
.outputAnnotationCount(item);
validateStreamEmitterMethod(item, outputAnnotationCount,
methodAnnotatedOutboundName);
invokeSetupMethodOnToTargetChannel(item, k,
methodAnnotatedOutboundName);
}));
this.running = true;
}
}
finally {
this.lock.unlock();
}
}
@SuppressWarnings({ "rawtypes", "unchecked" })
private void invokeSetupMethodOnToTargetChannel(Method method, Object bean,
String outboundName) {
Object[] arguments = new Object[method.getParameterCount()];
Object targetBean = null;
for (int parameterIndex = 0; parameterIndex < arguments.length; parameterIndex++) {
MethodParameter methodParameter = new SynthesizingMethodParameter(method,
parameterIndex);
Class<?> parameterType = methodParameter.getParameterType();
Object targetReferenceValue = null;
if (methodParameter.hasParameterAnnotation(Output.class)) {
targetReferenceValue = AnnotationUtils
.getValue(methodParameter.getParameterAnnotation(Output.class));
}
else if (arguments.length == 1 && StringUtils.hasText(outboundName)) {
targetReferenceValue = outboundName;
}
if (targetReferenceValue != null) {
targetBean = this.applicationContext
.getBean((String) targetReferenceValue);
for (StreamListenerParameterAdapter<?, Object> streamListenerParameterAdapter : this.parameterAdapters) {
if (streamListenerParameterAdapter.supports(targetBean.getClass(),
methodParameter)) {
arguments[parameterIndex] = streamListenerParameterAdapter
.adapt(targetBean, methodParameter);
if (arguments[parameterIndex] instanceof FluxSender) {
this.closeableFluxResources
.add((FluxSender) arguments[parameterIndex]);
}
break;
}
}
Assert.notNull(arguments[parameterIndex],
"Cannot convert argument " + parameterIndex + " of " + method
+ "from " + targetBean.getClass() + " to "
+ parameterType);
}
else {
throw new IllegalStateException(
StreamEmitterErrorMessages.ATLEAST_ONE_OUTPUT);
}
}
Object result;
try {
result = method.invoke(bean, arguments);
}
catch (Exception e) {
throw new BeanInitializationException(
"Cannot setup StreamEmitter for " + method, e);
}
if (!Void.TYPE.equals(method.getReturnType())) {
if (targetBean == null) {
targetBean = this.applicationContext.getBean(outboundName);
}
boolean streamListenerResultAdapterFound = false;
for (StreamListenerResultAdapter streamListenerResultAdapter : this.resultAdapters) {
if (streamListenerResultAdapter.supports(result.getClass(),
targetBean.getClass())) {
Closeable fluxDisposable = streamListenerResultAdapter.adapt(result,
targetBean);
this.closeableFluxResources.add(fluxDisposable);
streamListenerResultAdapterFound = true;
break;
}
}
Assert.state(streamListenerResultAdapterFound,
StreamEmitterErrorMessages.CANNOT_CONVERT_RETURN_TYPE_TO_ANY_AVAILABLE_RESULT_ADAPTERS);
}
}
@Override
public boolean isAutoStartup() {
return true;
}
@Override
public void stop(Runnable callback) {
stop();
if (callback != null) {
callback.run();
}
}
@Override
public void stop() {
try {
this.lock.lock();
if (this.running) {
for (Closeable closeable : this.closeableFluxResources) {
try {
closeable.close();
}
catch (IOException e) {
log.error("Error closing reactive source", e);
}
}
this.running = false;
}
}
finally {
this.lock.unlock();
}
}
@Override
public boolean isRunning() {
return this.running;
}
@Override
public int getPhase() {
return 0;
}
}

47
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitterErrorMessages.java
View File

@@ -1,47 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import org.springframework.cloud.stream.binding.StreamAnnotationErrorMessages;
/**
* @author Soby Chacko
* @since 1.3.0
*/
abstract class StreamEmitterErrorMessages extends StreamAnnotationErrorMessages {
static final String INVALID_OUTPUT_METHOD_PARAMETERS = "@Output annotations are not permitted on "
+ "method parameters while using the @StreamEmitter and a method-level output specification";
static final String NO_OUTPUT_SPECIFIED = "No method level or parameter level @Output annotations are detected. "
+ "@StreamEmitter requires a method or parameter level @Output annotation.";
// @checkstyle:off
static final String CANNOT_CONVERT_RETURN_TYPE_TO_ANY_AVAILABLE_RESULT_ADAPTERS = "No suitable adapters are found that can convert the return type";
// @checkstyle:on
private static final String PREFIX = "A method annotated with @StreamEmitter ";
static final String RETURN_TYPE_NO_OUTBOUND_SPECIFIED = PREFIX
+ "having a return type should also have an outbound target specified at the method level.";
static final String RETURN_TYPE_METHOD_ARGUMENTS = PREFIX
+ "having a return type should not have any method arguments";
static final String OUTPUT_ANNOTATION_MISSING_ON_METHOD_PARAMETERS_VOID_RETURN_TYPE = PREFIX
+ "and void return type without method level @Output annotation requires @Output on each of the method parameter";
static final String INPUT_ANNOTATIONS_ARE_NOT_ALLOWED = PREFIX
+ "cannot contain @Input annotations";
}

2
spring-cloud-stream-reactive/src/main/resources/META-INF/spring.factories
View File

@@ -1,2 +0,0 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.stream.reactive.ReactiveSupportAutoConfiguration

44
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/converter/TestApplicationJsonMessageMarshallingConverter.java
View File

@@ -1,44 +0,0 @@
/*
* Copyright 2019-2019 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.converter;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.MapperFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cloud.stream.reactive.MessageChannelToInputFluxParameterAdapterTests;
/**
* This class exists because the {@link ApplicationJsonMessageMarshallingConverter} is package-private scoped,
* so it can't be used directly in, e.g. the {@link MessageChannelToInputFluxParameterAdapterTests} class.
* In that case, we are just extending it and making it public (but only for tests).
*
* @author Ryan Dunckel
*/
public class TestApplicationJsonMessageMarshallingConverter
extends ApplicationJsonMessageMarshallingConverter {
public TestApplicationJsonMessageMarshallingConverter() {
this(new ObjectMapper().configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false)
.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false));
}
public TestApplicationJsonMessageMarshallingConverter(ObjectMapper objectMapper) {
super(objectMapper);
}
}

233
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapterTests.java
View File

@@ -1,233 +0,0 @@
/*
* Copyright 2016-2019 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.reactive;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.cloud.stream.converter.TestApplicationJsonMessageMarshallingConverter;
import org.springframework.core.MethodParameter;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.ReflectionUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ryan Dunckel
*/
public class MessageChannelToInputFluxParameterAdapterTests {
@Test
public void testWrapperFluxSupportsMultipleSubscriptions() throws Exception {
List<String> results = Collections.synchronizedList(new ArrayList<>());
CountDownLatch latch = new CountDownLatch(4);
final MessageChannelToInputFluxParameterAdapter messageChannelToInputFluxParameterAdapter;
messageChannelToInputFluxParameterAdapter = new MessageChannelToInputFluxParameterAdapter(
new CompositeMessageConverter(
Collections.singleton(new MappingJackson2MessageConverter())));
final Method processMethod = ReflectionUtils.findMethod(
MessageChannelToInputFluxParameterAdapterTests.class, "process",
Flux.class);
final DirectChannel adaptedChannel = new DirectChannel();
@SuppressWarnings("unchecked")
final Flux<Message<?>> adapterFlux = (Flux<Message<?>>) messageChannelToInputFluxParameterAdapter
.adapt(adaptedChannel, new MethodParameter(processMethod, 0));
String uuid1 = UUID.randomUUID().toString();
String uuid2 = UUID.randomUUID().toString();
adapterFlux.map(m -> m.getPayload() + uuid1).subscribe(s -> {
results.add(s);
latch.countDown();
});
adapterFlux.map(m -> m.getPayload() + uuid2).subscribe(s -> {
results.add(s);
latch.countDown();
});
adaptedChannel.send(MessageBuilder.withPayload("A").build());
adaptedChannel.send(MessageBuilder.withPayload("B").build());
assertThat(latch.await(5000, TimeUnit.MILLISECONDS)).isTrue();
assertThat(results).containsExactlyInAnyOrder("A" + uuid1, "B" + uuid1,
"A" + uuid2, "B" + uuid2);
}
@Test
public void testAdapterConvertsUsingConversionHint() {
CompositeMessageConverter messageConverter = new CompositeMessageConverter(
Collections
.singleton(new TestApplicationJsonMessageMarshallingConverter()));
MessageChannelToInputFluxParameterAdapter adapter = new MessageChannelToInputFluxParameterAdapter(
messageConverter);
Method processMethod = ReflectionUtils.findMethod(
MessageChannelToInputFluxParameterAdapterTests.class, "processNestedGenericFlux",
Flux.class);
DirectChannel adaptedChannel = new DirectChannel();
@SuppressWarnings("unchecked")
Flux<FirstLevelWrapper<SecondLevelWrapper<String>>> adapterFlux = (Flux<FirstLevelWrapper<SecondLevelWrapper<String>>>) adapter
.adapt(adaptedChannel, new MethodParameter(processMethod, 0));
SecondLevelWrapper<String> expected2 = new SecondLevelWrapper<>();
expected2.setName("name");
expected2.setData("data");
FirstLevelWrapper<SecondLevelWrapper<String>> expected1 = new FirstLevelWrapper<>();
expected1.setId(1);
expected1.setData(expected2);
StepVerifier.create(adapterFlux).then(() -> {
adaptedChannel.send(MessageBuilder.withPayload(
"{ \"id\": 1, \"data\": { \"name\": \"name\", \"data\": \"data\" } }")
.build());
}).expectNext(expected1).thenCancel().verify();
}
public void process(Flux<Message<?>> message) {
// do nothing - we just reference this method from the test
}
public void processNestedGenericFlux(Flux<FirstLevelWrapper<SecondLevelWrapper<String>>> items) {
// do nothing - we just reference this method from the test
}
static class FirstLevelWrapper<T> {
private long id;
private T data;
long getId() {
return id;
}
void setId(long id) {
this.id = id;
}
T getData() {
return data;
}
public void setData(T data) {
this.data = data;
}
@Override
public int hashCode() {
return Objects.hash(data, id);
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj == null) {
return false;
}
if (!(obj instanceof FirstLevelWrapper)) {
return false;
}
@SuppressWarnings("unchecked")
FirstLevelWrapper<T> other = (FirstLevelWrapper<T>) obj;
return Objects.equals(data, other.data) && id == other.id;
}
@Override
public String toString() {
return "FirstLevelWrapper [id=" + id + ", data=" + data + "]";
}
}
static class SecondLevelWrapper<T> {
private String name;
private T data;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public T getData() {
return data;
}
public void setData(T data) {
this.data = data;
}
@Override
public int hashCode() {
return Objects.hash(data, name);
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj == null) {
return false;
}
if (!(obj instanceof SecondLevelWrapper)) {
return false;
}
@SuppressWarnings("unchecked")
SecondLevelWrapper<T> other = (SecondLevelWrapper<T>) obj;
return Objects.equals(data, other.data) && Objects.equals(name, other.name);
}
@Override
public String toString() {
return "SecondLevelWrapper [name=" + name + ", data=" + data + "]";
}
}
}

347
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterBasicTests.java
View File

@@ -1,347 +0,0 @@
/*
* Copyright 2017-2019 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.reactive;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Soby Chacko
* @author Artem Bilan
* @author Vinicius Carvalho
* @author Oleg Zhurakousky
*/
public class StreamEmitterBasicTests {
private static void receiveAndValidate(ConfigurableApplicationContext context)
throws InterruptedException {
Source source = context.getBean(Source.class);
MessageCollector messageCollector = context.getBean(MessageCollector.class);
List<String> messages = new ArrayList<>();
for (int i = 0; i < 1000; i++) {
messages.add((String) messageCollector.forChannel(source.output())
.poll(5000, TimeUnit.MILLISECONDS).getPayload());
}
for (int i = 0; i < 1000; i++) {
assertThat(new String(messages.get(i))).isEqualTo("HELLO WORLD!!" + i);
}
}
private static void receiveAndValidateMultipleOutputs(
ConfigurableApplicationContext context) throws InterruptedException {
TestMultiOutboundChannels source = context
.getBean(TestMultiOutboundChannels.class);
MessageCollector messageCollector = context.getBean(MessageCollector.class);
List<String> messages = new ArrayList<>();
assertMessages(source.output1(), messageCollector, messages);
messages.clear();
assertMessages(source.output2(), messageCollector, messages);
messages.clear();
assertMessages(source.output3(), messageCollector, messages);
messages.clear();
}
private static void receiveAndValidateMultiStreamEmittersInSameContext(
ConfigurableApplicationContext context1) throws InterruptedException {
TestMultiOutboundChannels source1 = context1
.getBean(TestMultiOutboundChannels.class);
MessageCollector messageCollector = context1.getBean(MessageCollector.class);
List<String> messages = new ArrayList<>();
assertMessagesX(source1.output1(), messageCollector, messages);
messages.clear();
assertMessagesY(source1.output2(), messageCollector, messages);
messages.clear();
}
private static void assertMessages(MessageChannel channel,
MessageCollector messageCollector, List<String> messages)
throws InterruptedException {
for (int i = 0; i < 1000; i++) {
messages.add((String) messageCollector.forChannel(channel)
.poll(5000, TimeUnit.MILLISECONDS).getPayload());
}
for (int i = 0; i < 1000; i++) {
assertThat(new String(messages.get(i))).isEqualTo("Hello World!!" + i);
}
}
private static void assertMessagesX(MessageChannel channel,
MessageCollector messageCollector, List<String> messages)
throws InterruptedException {
for (int i = 0; i < 1000; i++) {
messages.add((String) messageCollector.forChannel(channel)
.poll(5000, TimeUnit.MILLISECONDS).getPayload());
}
for (int i = 0; i < 1000; i++) {
assertThat(new String(messages.get(i))).isEqualTo("Hello World!!" + i);
}
}
private static void assertMessagesY(MessageChannel channel,
MessageCollector messageCollector, List<String> messages)
throws InterruptedException {
for (int i = 0; i < 1000; i++) {
messages.add((String) messageCollector.forChannel(channel)
.poll(5000, TimeUnit.MILLISECONDS).getPayload());
}
for (int i = 0; i < 1000; i++) {
assertThat(new String(messages.get(i))).isEqualTo("Hello FooBar!!" + i);
}
}
@Test
public void testFluxReturnAndOutputMethodLevel() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(
TestFluxReturnAndOutputMethodLevel.class, "--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
receiveAndValidate(context);
context.close();
}
@Test
public void testVoidReturnAndOutputMethodParameter() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(
TestVoidReturnAndOutputMethodParameter.class, "--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
receiveAndValidate(context);
context.close();
}
@Test
public void testVoidReturnAndOutputAtMethodLevel() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(
TestVoidReturnAndOutputAtMethodLevel.class, "--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
receiveAndValidate(context);
context.close();
}
@Test
public void testVoidReturnAndMultipleOutputMethodParameters() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(
TestVoidReturnAndMultipleOutputMethodParameters.class, "--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain",
"--spring.cloud.stream.bindings.output1.contentType=text/plain",
"--spring.cloud.stream.bindings.output2.contentType=text/plain",
"--spring.cloud.stream.bindings.output3.contentType=text/plain");
receiveAndValidateMultipleOutputs(context);
context.close();
}
@Test
public void testMultipleStreamEmitterMethods() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(
TestMultipleStreamEmitterMethods.class, "--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain",
"--spring.cloud.stream.bindings.output1.contentType=text/plain",
"--spring.cloud.stream.bindings.output2.contentType=text/plain",
"--spring.cloud.stream.bindings.output3.contentType=text/plain");
receiveAndValidateMultipleOutputs(context);
context.close();
}
@Test
public void testSameAppContextWithMultipleStreamEmitters() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(
TestSameAppContextWithMultipleStreamEmitters.class, "--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain",
"--spring.cloud.stream.bindings.output1.contentType=text/plain",
"--spring.cloud.stream.bindings.output2.contentType=text/plain",
"--spring.cloud.stream.bindings.output3.contentType=text/plain");
receiveAndValidateMultiStreamEmittersInSameContext(context);
context.close();
}
interface TestMultiOutboundChannels {
String OUTPUT1 = "output1";
String OUTPUT2 = "output2";
String OUTPUT3 = "output3";
@Output(TestMultiOutboundChannels.OUTPUT1)
MessageChannel output1();
@Output(TestMultiOutboundChannels.OUTPUT2)
MessageChannel output2();
@Output(TestMultiOutboundChannels.OUTPUT3)
MessageChannel output3();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestFluxReturnAndOutputMethodLevel {
@StreamEmitter
@Output(Source.OUTPUT)
@Bean
public Publisher<Message<String>> emit() {
AtomicInteger atomicInteger = new AtomicInteger();
return IntegrationFlows
.from((Supplier<?>) () -> new GenericMessage<>(
"Hello World!!" + atomicInteger.getAndIncrement()),
e -> e.poller(p -> p.fixedDelay(1)))
.<String, String>transform(String::toUpperCase).toReactivePublisher();
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestVoidReturnAndOutputMethodParameter {
@StreamEmitter
public void emit(@Output(Source.OUTPUT) FluxSender output) {
output.send(Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l)
.map(String::toUpperCase));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestVoidReturnAndOutputAtMethodLevel {
@StreamEmitter
@Output(Source.OUTPUT)
public void emit(FluxSender output) {
output.send(Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l)
.map(String::toUpperCase));
}
}
@EnableBinding(TestMultiOutboundChannels.class)
@EnableAutoConfiguration
public static class TestVoidReturnAndMultipleOutputMethodParameters {
@StreamEmitter
public void emit(@Output(TestMultiOutboundChannels.OUTPUT1) FluxSender output1,
@Output(TestMultiOutboundChannels.OUTPUT2) FluxSender output2,
@Output(TestMultiOutboundChannels.OUTPUT3) FluxSender output3) {
output1.send(
Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
output2.send(
Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
output3.send(
Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
}
}
@EnableBinding(TestMultiOutboundChannels.class)
@EnableAutoConfiguration
public static class TestMultipleStreamEmitterMethods {
@StreamEmitter
@Output(TestMultiOutboundChannels.OUTPUT1)
public Flux<String> emit1() {
return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
}
@StreamEmitter
@Output(TestMultiOutboundChannels.OUTPUT2)
public Flux<String> emit2() {
return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
}
@StreamEmitter
public void emit3(@Output(TestMultiOutboundChannels.OUTPUT3) FluxSender outputX) {
outputX.send(
Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
}
}
@EnableBinding(TestMultiOutboundChannels.class)
@EnableAutoConfiguration
public static class TestSameAppContextWithMultipleStreamEmitters {
@Bean
public Foo foo() {
return new Foo();
}
@Bean
public Bar bar() {
return new Bar();
}
static class Foo {
@StreamEmitter
@Output(TestMultiOutboundChannels.OUTPUT1)
public Flux<String> emit1() {
return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
}
}
static class Bar {
@StreamEmitter
@Output(TestMultiOutboundChannels.OUTPUT2)
public Flux<String> emit2() {
return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello FooBar!!" + l);
}
}
}
}

323
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterValidationTests.java
View File

@@ -1,323 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.time.Duration;
import org.junit.Test;
import reactor.core.publisher.Flux;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.springframework.cloud.stream.binding.StreamAnnotationErrorMessages.ATLEAST_ONE_OUTPUT;
import static org.springframework.cloud.stream.binding.StreamAnnotationErrorMessages.INVALID_OUTBOUND_NAME;
import static org.springframework.cloud.stream.reactive.StreamEmitterErrorMessages.CANNOT_CONVERT_RETURN_TYPE_TO_ANY_AVAILABLE_RESULT_ADAPTERS;
import static org.springframework.cloud.stream.reactive.StreamEmitterErrorMessages.INPUT_ANNOTATIONS_ARE_NOT_ALLOWED;
import static org.springframework.cloud.stream.reactive.StreamEmitterErrorMessages.INVALID_OUTPUT_METHOD_PARAMETERS;
import static org.springframework.cloud.stream.reactive.StreamEmitterErrorMessages.NO_OUTPUT_SPECIFIED;
import static org.springframework.cloud.stream.reactive.StreamEmitterErrorMessages.OUTPUT_ANNOTATION_MISSING_ON_METHOD_PARAMETERS_VOID_RETURN_TYPE;
import static org.springframework.cloud.stream.reactive.StreamEmitterErrorMessages.RETURN_TYPE_METHOD_ARGUMENTS;
import static org.springframework.cloud.stream.reactive.StreamEmitterErrorMessages.RETURN_TYPE_NO_OUTBOUND_SPECIFIED;
/**
* @author Soby Chacko
* @author Vinicius Carvalho
*/
public class StreamEmitterValidationTests {
@Test
public void testOutputAsMethodandMethodParameter() {
try {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(TestOutputAsMethodandMethodParameter.class);
context.refresh();
context.close();
fail("Expected exception: " + INVALID_OUTPUT_METHOD_PARAMETERS);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(INVALID_OUTPUT_METHOD_PARAMETERS);
}
}
@Test
public void testFluxReturnTypeNoOutputGiven() {
try {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(TestFluxReturnTypeNoOutputGiven.class);
context.refresh();
context.close();
fail("Expected exception: " + NO_OUTPUT_SPECIFIED);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(NO_OUTPUT_SPECIFIED);
}
}
@Test
public void testVoidReturnTypeNoOutputGiven() {
try {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(TestVoidReturnTypeNoOutputGiven.class);
context.refresh();
context.close();
fail("Expected exception: " + NO_OUTPUT_SPECIFIED);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(NO_OUTPUT_SPECIFIED);
}
}
@Test
public void testNonVoidReturnButOutputAsMethodParameter() {
try {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(TestNonVoidReturnButOutputAsMethodParameter.class);
context.refresh();
context.close();
fail("Expected exception: " + RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
}
}
@Test
public void testNonVoidReturnButMethodArguments() {
try {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(TestNonVoidReturnButMethodArguments.class);
context.refresh();
context.close();
fail("Expected exception: " + RETURN_TYPE_METHOD_ARGUMENTS);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(RETURN_TYPE_METHOD_ARGUMENTS);
}
}
@Test
public void testVoidReturnTypeMultipleMethodParametersWithOneMissingOutput() {
try {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(
TestVoidReturnTypeMultipleMethodParametersWithOneMissingOutput.class);
context.refresh();
context.close();
fail("Expected exception: "
+ OUTPUT_ANNOTATION_MISSING_ON_METHOD_PARAMETERS_VOID_RETURN_TYPE);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(
OUTPUT_ANNOTATION_MISSING_ON_METHOD_PARAMETERS_VOID_RETURN_TYPE);
}
}
@Test
public void testOutputAtCorrectLocationButNameMissing1() {
try {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(TestOutputAtCorrectLocationButNameMissing1.class);
context.refresh();
context.close();
fail("Expected exception: " + ATLEAST_ONE_OUTPUT);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(ATLEAST_ONE_OUTPUT);
}
}
@Test
public void testOutputAtCorrectLocationButNameMissing2() {
try {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(TestOutputAtCorrectLocationButNameMissing2.class);
context.refresh();
context.close();
fail("Expected exception: " + INVALID_OUTBOUND_NAME);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(INVALID_OUTBOUND_NAME);
}
}
@Test
public void testInputAnnotationsAreNotPermitted() {
try {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(TestInputAnnotationsAreNotPermitted.class);
context.refresh();
context.close();
fail("Expected exception: " + INPUT_ANNOTATIONS_ARE_NOT_ALLOWED);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(INPUT_ANNOTATIONS_ARE_NOT_ALLOWED);
}
}
@Test
public void testReturnTypeNotSupported() {
try {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
context.register(TestReturnTypeNotSupported.class);
context.refresh();
context.close();
fail("Expected exception: "
+ CANNOT_CONVERT_RETURN_TYPE_TO_ANY_AVAILABLE_RESULT_ADAPTERS);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(
CANNOT_CONVERT_RETURN_TYPE_TO_ANY_AVAILABLE_RESULT_ADAPTERS);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestOutputAsMethodandMethodParameter {
@StreamEmitter
@Output(Source.OUTPUT)
public void receive(@Output(Source.OUTPUT) FluxSender output) {
output.send(
Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestFluxReturnTypeNoOutputGiven {
@StreamEmitter
public Flux<String> emit() {
return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestVoidReturnTypeNoOutputGiven {
@StreamEmitter
public void emit(FluxSender output) {
output.send(
Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestNonVoidReturnButOutputAsMethodParameter {
@StreamEmitter
public Flux<String> emit(@Output(Source.OUTPUT) FluxSender output) {
return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestNonVoidReturnButMethodArguments {
@StreamEmitter
@Output(Source.OUTPUT)
public Flux<String> receive(FluxSender output) {
return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
}
}
@EnableBinding(StreamEmitterBasicTests.TestMultiOutboundChannels.class)
@EnableAutoConfiguration
public static class TestVoidReturnTypeMultipleMethodParametersWithOneMissingOutput {
@StreamEmitter
public void emit(
@Output(StreamEmitterBasicTests.TestMultiOutboundChannels.OUTPUT1) FluxSender output1,
@Output(StreamEmitterBasicTests.TestMultiOutboundChannels.OUTPUT2) FluxSender output2,
FluxSender output3) {
output1.send(
Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
output2.send(
Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
output3.send(
Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestOutputAtCorrectLocationButNameMissing1 {
@StreamEmitter
@Output("")
public void receive(FluxSender output) {
output.send(
Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
}
}
@EnableBinding(StreamEmitterBasicTests.TestMultiOutboundChannels.class)
@EnableAutoConfiguration
public static class TestOutputAtCorrectLocationButNameMissing2 {
@StreamEmitter
public void emit(@Output("") FluxSender output1) {
output1.send(
Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputAnnotationsAreNotPermitted {
@StreamEmitter
@Output(Source.OUTPUT)
@Input(Processor.INPUT)
public Flux<String> emit() {
return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestReturnTypeNotSupported {
@StreamEmitter
@Output(Source.OUTPUT)
public String emit() {
return "hello";
}
}
}

126
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerGenericFluxInputOutputArgsWithMessageTests.java
View File

@@ -1,126 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Ignore;
import org.junit.Test;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM;
/**
* @author Ilayaperumal Gopinathan
* @author Vinicius Carvalho
* @author Oleg Zhurakousky
*/
@SuppressWarnings("unchecked")
@Ignore
public class StreamListenerGenericFluxInputOutputArgsWithMessageTests {
private static void sendMessageAndValidate(ConfigurableApplicationContext context)
throws InterruptedException {
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload)
.setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<String> result = (Message<String>) messageCollector
.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@Test
public void testGenericFluxInputOutputArgsWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(
TestGenericStringFluxInputOutputArgsWithMessageImpl1.class,
"--server.port=0", "--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testInvalidInputValueWithOutputMethodParameters() {
try {
SpringApplication.run(
TestGenericStringFluxInputOutputArgsWithMessageImpl2.class,
"--server.port=0", "--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
fail("Expected exception: " + INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
catch (Exception e) {
assertThat(e.getMessage())
.contains(INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
}
public static class TestGenericStringFluxInputOutputArgsWithMessageImpl1
extends TestGenericFluxInputOutputArgsWithMessage1<String> {
}
public static class TestGenericStringFluxInputOutputArgsWithMessageImpl2
extends TestGenericFluxInputOutputArgsWithMessage2<String> {
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestGenericFluxInputOutputArgsWithMessage1<A> {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<A> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder
.withPayload((A) m.toString().toUpperCase()).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestGenericFluxInputOutputArgsWithMessage2<A> {
@StreamListener(Processor.INPUT)
public void receive(Flux<A> input, @Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder
.withPayload((A) m.toString().toUpperCase()).build()));
}
}
}

79
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerInterruptionTests.java
View File

@@ -1,79 +0,0 @@
/*
* Copyright 2016-2019 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.reactive;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Test validating that a fix for
* <a href="https://github.com/reactor/reactor-core/issues/159"/> is present.
*
* @author Marius Bogoevici
*/
public class StreamListenerInterruptionTests {
@Test
public void testSubscribersNotInterrupted() throws Exception {
ConfigurableApplicationContext context = SpringApplication
.run(TestTimeWindows.class, "--server.port=0");
Sink sink = context.getBean(Sink.class);
TestTimeWindows testTimeWindows = context.getBean(TestTimeWindows.class);
sink.input().send(MessageBuilder.withPayload("hello1").build());
sink.input().send(MessageBuilder.withPayload("hello2").build());
sink.input().send(MessageBuilder.withPayload("hello3").build());
assertThat(testTimeWindows.latch.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(testTimeWindows.interruptionState).isNotNull();
assertThat(testTimeWindows.interruptionState).isFalse();
context.close();
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
public static class TestTimeWindows {
public CountDownLatch latch = new CountDownLatch(1);
public Boolean interruptionState;
@StreamListener
public void receive(@Input(Sink.INPUT) Flux<String> input) {
input.window(Duration.ofMillis(500), Duration.ofMillis(100))
.flatMap(w -> w.reduce("", (x, y) -> x + y)).subscribe(x -> {
this.interruptionState = Thread.currentThread().isInterrupted();
this.latch.countDown();
});
}
}
}

98
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java
View File

@@ -1,98 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.util.Collection;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveInputOutputArgsTests {
private Class<?> configClass;
public StreamListenerReactiveInputOutputArgsTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection<?> InputConfigs() {
return Collections.singletonList(ReactorTestInputOutputArgs.class);
}
@SuppressWarnings("unchecked")
private static void sendMessageAndValidate(ConfigurableApplicationContext context)
throws InterruptedException {
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload)
.setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<String> result = (Message<String>) messageCollector
.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@Test
public void testInputOutputArgs() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
"--server.port=0", "--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
sendMessageAndValidate(context);
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestInputOutputArgs {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<String> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> m.toUpperCase()));
}
}
}

99
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java
View File

@@ -1,99 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.util.Collection;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveInputOutputArgsWithMessageTests {
private Class<?> configClass;
public StreamListenerReactiveInputOutputArgsWithMessageTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection<?> InputConfigs() {
return Collections.singletonList(ReactorTestInputOutputArgsWithMessage.class);
}
@SuppressWarnings("unchecked")
private static void sendMessageAndValidate(ConfigurableApplicationContext context)
throws InterruptedException {
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload)
.setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<String> result = (Message<String>) messageCollector
.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@Test
public void testInputOutputArgs() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
"--server.port=0", "--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
sendMessageAndValidate(context);
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestInputOutputArgsWithMessage {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<?>> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder
.withPayload(m.getPayload().toString().toUpperCase()).build()));
}
}
}

116
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java
View File

@@ -1,116 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.util.Collection;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests {
private Class<?> configClass;
public StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests(
Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection<?> InputConfigs() {
return Collections
.singletonList(TestInputOutputArgsWithFluxSenderAndFailure.class);
}
@SuppressWarnings("unchecked")
private static void sendMessageAndValidate(ConfigurableApplicationContext context)
throws InterruptedException {
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload)
.setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<String> result = (Message<String>) messageCollector
.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
private static void sendFailingMessage(ConfigurableApplicationContext context)
throws InterruptedException {
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("fail")
.setHeader("contentType", "text/plain").build());
}
@Test
public void testInputOutputArgsWithFluxSenderAndFailure() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
"--server.port=0", "--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
sendMessageAndValidate(context);
sendFailingMessage(context);
sendMessageAndValidate(context);
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgsWithFluxSenderAndFailure {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<String>> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> m.getPayload().toString()).map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
}).map(o -> MessageBuilder.withPayload(o).build()));
}
}
}

102
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java
View File

@@ -1,102 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.util.Collection;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveInputOutputArgsWithSenderTests {
private Class<?> configClass;
public StreamListenerReactiveInputOutputArgsWithSenderTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection<?> InputConfigs() {
return Collections.singletonList(ReactorTestInputOutputArgsWithFluxSender.class);
}
@SuppressWarnings("unchecked")
private static void sendMessageAndValidate(ConfigurableApplicationContext context)
throws InterruptedException {
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload)
.setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<String> result = (Message<String>) messageCollector
.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@Test
public void testInputOutputArgsWithFluxSender() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
"--server.port=0", "--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
// send multiple message
sendMessageAndValidate(context);
sendMessageAndValidate(context);
sendMessageAndValidate(context);
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestInputOutputArgsWithFluxSender {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<String>> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> m.getPayload().toString().toUpperCase())
.map(o -> MessageBuilder.withPayload(o).build()));
}
}
}

89
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodTests.java
View File

@@ -1,89 +0,0 @@
/*
* Copyright 2016-2019 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.reactive;
import org.junit.Test;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.RETURN_TYPE_NO_OUTBOUND_SPECIFIED;
/**
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerReactiveMethodTests {
@Test
public void testReactiveInvalidInputValueWithOutputMethodParameters() {
try {
SpringApplication.run(ReactorTestInputOutputArgs.class, "--server.port=0");
fail("IllegalArgumentException should have been thrown");
}
catch (Exception e) {
assertThat(e.getMessage())
.contains(INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
}
@Test
public void testMethodReturnTypeWithNoOutboundSpecified() {
try {
SpringApplication.run(ReactorTestReturn5.class, "--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
fail("Exception expected: " + RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestInputOutputArgs {
@StreamListener(Processor.INPUT)
public void receive(Flux<String> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> m.toUpperCase()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturn5 {
@StreamListener
public Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> m.toUpperCase());
}
}
}

138
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java
View File

@@ -1,138 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveMethodWithReturnTypeTests {
private Class<?> configClass;
public StreamListenerReactiveMethodWithReturnTypeTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection<?> InputConfigs() {
return Arrays.asList(ReactorTestReturn1.class, ReactorTestReturn2.class,
ReactorTestReturn3.class, ReactorTestReturn4.class);
}
@SuppressWarnings("unchecked")
private static void sendMessageAndValidate(ConfigurableApplicationContext context)
throws InterruptedException {
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload)
.setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<String> result = (Message<String>) messageCollector
.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@Test
public void testReturn() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
"--server.port=0", "--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
sendMessageAndValidate(context);
sendMessageAndValidate(context);
sendMessageAndValidate(context);
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturn1 {
@StreamListener
public @Output(Processor.OUTPUT) Flux<String> receive(
@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturn2 {
@StreamListener(Processor.INPUT)
@Output(Processor.OUTPUT)
public Flux<String> receive(Flux<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturn3 {
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public Flux<String> receive(Flux<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturn4 {
@StreamListener
@SendTo(Processor.OUTPUT)
public Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> m.toUpperCase());
}
}
}

173
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java
View File

@@ -1,173 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveReturnWithFailureTests {
private Class<?> configClass;
public StreamListenerReactiveReturnWithFailureTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection<?> InputConfigs() {
return Arrays.asList(ReactorTestReturnWithFailure1.class,
ReactorTestReturnWithFailure2.class, ReactorTestReturnWithFailure3.class,
ReactorTestReturnWithFailure4.class);
}
@SuppressWarnings("unchecked")
private static void sendMessageAndValidate(ConfigurableApplicationContext context)
throws InterruptedException {
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload)
.setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<String> result = (Message<String>) messageCollector
.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
private static void sendFailingMessage(ConfigurableApplicationContext context) {
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("fail")
.setHeader("contentType", "text/plain").build());
}
@Test
public void testReturnWithFailure() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
"--server.port=0", "--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
sendMessageAndValidate(context);
sendFailingMessage(context);
sendMessageAndValidate(context);
sendFailingMessage(context);
sendMessageAndValidate(context);
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithFailure1 {
@StreamListener
public @Output(Processor.OUTPUT) Flux<String> receive(
@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithFailure2 {
@StreamListener(Processor.INPUT)
public @Output(Processor.OUTPUT) Flux<String> receive(Flux<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithFailure3 {
@StreamListener(Processor.INPUT)
public @SendTo(Processor.OUTPUT) Flux<String> receive(Flux<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithFailure4 {
@StreamListener
public @SendTo(Processor.OUTPUT) Flux<String> receive(
@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
}

137
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java
View File

@@ -1,137 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveReturnWithMessageTests {
private Class<?> configClass;
public StreamListenerReactiveReturnWithMessageTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection<?> InputConfigs() {
return Arrays.asList(ReactorTestReturnWithMessage1.class,
ReactorTestReturnWithMessage2.class, ReactorTestReturnWithMessage3.class,
ReactorTestReturnWithMessage4.class);
}
@SuppressWarnings("unchecked")
private static void sendMessageAndValidate(ConfigurableApplicationContext context)
throws InterruptedException {
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload)
.setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<String> result = (Message<String>) messageCollector
.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@Test
public void testReturnWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
"--server.port=0", "--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.contentType=text/plain",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
sendMessageAndValidate(context);
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithMessage1 {
@StreamListener
public @Output(Processor.OUTPUT) Flux<String> receive(
@Input(Processor.INPUT) Flux<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithMessage2 {
@StreamListener(Processor.INPUT)
public @Output(Processor.OUTPUT) Flux<String> receive(
Flux<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithMessage3 {
@StreamListener(Processor.INPUT)
public @SendTo(Processor.OUTPUT) Flux<String> receive(
Flux<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithMessage4 {
@StreamListener
public @SendTo(Processor.OUTPUT) Flux<String> receive(
@Input(Processor.INPUT) Flux<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
}

165
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java
View File

@@ -1,165 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveReturnWithPojoTests {
private Class<?> configClass;
private ObjectMapper mapper = new ObjectMapper();
public StreamListenerReactiveReturnWithPojoTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection<?> InputConfigs() {
return Arrays.asList(ReactorTestReturnWithPojo1.class,
ReactorTestReturnWithPojo2.class, ReactorTestReturnWithPojo3.class,
ReactorTestReturnWithPojo4.class);
}
@Test
@SuppressWarnings("unchecked")
public void testReturnWithPojo() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
"--server.port=0", "--spring.jmx.enabled=false");
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("{\"message\":\"helloPojo\"}")
.setHeader("contentType", "application/json").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<String> result = (Message<String>) messageCollector
.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
BarPojo barPojo = this.mapper.readValue(result.getPayload(), BarPojo.class);
assertThat(barPojo.getBarMessage()).isEqualTo("helloPojo");
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithPojo1 {
@StreamListener
public @Output(Processor.OUTPUT) Flux<BarPojo> receive(
@Input(Processor.INPUT) Flux<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithPojo2 {
@StreamListener(Processor.INPUT)
public @Output(Processor.OUTPUT) Flux<BarPojo> receive(Flux<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithPojo3 {
@StreamListener(Processor.INPUT)
public @SendTo(Processor.OUTPUT) Flux<BarPojo> receive(Flux<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithPojo4 {
@StreamListener
public @SendTo(Processor.OUTPUT) Flux<BarPojo> receive(
@Input(Processor.INPUT) Flux<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
public static class FooPojo {
private String message;
public String getMessage() {
return this.message;
}
public void setMessage(String message) {
this.message = message;
}
}
public static class BarPojo {
private String barMessage;
@JsonCreator
public BarPojo(@JsonProperty("barMessage") String barMessage) {
this.barMessage = barMessage;
}
public String getBarMessage() {
return this.barMessage;
}
public void setBarMessage(String barMessage) {
this.barMessage = barMessage;
}
}
}

135
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerWildCardFluxInputOutputArgsWithMessageTests.java
View File

@@ -1,135 +0,0 @@
/*
* Copyright 2016-2017 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.reactive;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM;
/**
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
*/
public class StreamListenerWildCardFluxInputOutputArgsWithMessageTests {
@SuppressWarnings("unchecked")
private static void sendMessageAndValidate(ConfigurableApplicationContext context)
throws InterruptedException {
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload)
.setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<String> result = (Message<String>) messageCollector
.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@Test
public void testWildCardFluxInputOutputArgsWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(
TestWildCardFluxInputOutputArgsWithMessage1.class, "--server.port=0",
"--spring.cloud.stream.bindings.output.contentType=text/plain");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testInputAsStreamListenerAndOutputAsParameterUsage() {
try {
SpringApplication.run(TestWildCardFluxInputOutputArgsWithMessage2.class,
"--server.port=0");
fail("Expected exception: " + INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
catch (Exception e) {
assertThat(e.getMessage())
.contains(INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
}
@Test
public void testIncorrectUsage1() throws Exception {
try {
SpringApplication.run(TestWildCardFluxInputOutputArgsWithMessage3.class,
"--server.port=0");
fail("Expected exception: " + INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestWildCardFluxInputOutputArgsWithMessage1 {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<?> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(
m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestWildCardFluxInputOutputArgsWithMessage2 {
@StreamListener(Processor.INPUT)
public void receive(Flux<?> input, @Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(
m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestWildCardFluxInputOutputArgsWithMessage3 {
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public void receive(Flux<?> input, FluxSender output) {
output.send(input.map(
m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
}
}
}

10
spring-cloud-stream-reactive/src/test/resources/logback.xml
View File

@@ -1,10 +0,0 @@
<configuration>
<appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{ISO8601} %5p %t %c{2}:%L - %m%n</pattern>
</encoder>
</appender>
<root level="WARN">
<appender-ref ref="stdout"/>
</root>
</configuration>

BIN
spring-cloud-stream-schema/foodorder.avro
View File

Binary file not shown.