, 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;
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java
deleted file mode 100644
index 1edb14788..000000000
--- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java
+++ /dev/null
@@ -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();
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitter.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitter.java
deleted file mode 100644
index ec8a8d303..000000000
--- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitter.java
+++ /dev/null
@@ -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.
- *
- *
- * @StreamEmitter
- * @Output(Source.OUTPUT)
- * public Flux<String> emit() {
- * return Flux.intervalMillis(1000)
- * .map(l -> "Hello World!!");
- * }
- *
- *
- * 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.
- *
- *
- * @StreamEmitter
- * public void emit(@Output(Source.OUTPUT) FluxSender output) {
- * output.send(Flux.intervalMillis(1000)
- * .map(l -> "Hello World!!"));
- * }
- *
- *
- *
- * @StreamEmitter
- * @Output(Source.OUTPUT)
- * public void emit(FluxSender output) {
- * output.send(Flux.intervalMillis(1000)
- * .map(l -> "Hello World!!"));
- * }
- *
- *
- *
- * @StreamEmitter
- * public void emit(@Output("OUTPUT1") FluxSender output1,
- * @Output("OUTPUT2") FluxSender output2,
- * @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!!"));
- * }
- *
- *
- * @author Soby Chacko
- *
- * @since 1.3.0
- */
-@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
-@Retention(RetentionPolicy.RUNTIME)
-@Documented
-public @interface StreamEmitter {
-
-}
diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitterAnnotationBeanPostProcessor.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitterAnnotationBeanPostProcessor.java
deleted file mode 100644
index ea42d7575..000000000
--- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitterAnnotationBeanPostProcessor.java
+++ /dev/null
@@ -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 closeableFluxResources = new ArrayList<>();
-
- private final Lock lock = new ReentrantLock();
-
- @SuppressWarnings("rawtypes")
- private Collection parameterAdapters;
-
- @SuppressWarnings("rawtypes")
- private Collection resultAdapters;
-
- private ConfigurableApplicationContext applicationContext;
-
- private MultiValueMap 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 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;
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitterErrorMessages.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitterErrorMessages.java
deleted file mode 100644
index 94bdaca40..000000000
--- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/StreamEmitterErrorMessages.java
+++ /dev/null
@@ -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";
-
-}
diff --git a/spring-cloud-stream-reactive/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-reactive/src/main/resources/META-INF/spring.factories
deleted file mode 100644
index a8ce9d92c..000000000
--- a/spring-cloud-stream-reactive/src/main/resources/META-INF/spring.factories
+++ /dev/null
@@ -1,2 +0,0 @@
-org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-org.springframework.cloud.stream.reactive.ReactiveSupportAutoConfiguration
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/converter/TestApplicationJsonMessageMarshallingConverter.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/converter/TestApplicationJsonMessageMarshallingConverter.java
deleted file mode 100644
index 8be6d838d..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/converter/TestApplicationJsonMessageMarshallingConverter.java
+++ /dev/null
@@ -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);
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapterTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapterTests.java
deleted file mode 100644
index aa0f7d270..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapterTests.java
+++ /dev/null
@@ -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 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> adapterFlux = (Flux>) 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>> adapterFlux = (Flux>>) adapter
- .adapt(adaptedChannel, new MethodParameter(processMethod, 0));
-
- SecondLevelWrapper expected2 = new SecondLevelWrapper<>();
- expected2.setName("name");
- expected2.setData("data");
-
- FirstLevelWrapper> 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) {
- // do nothing - we just reference this method from the test
- }
-
- public void processNestedGenericFlux(Flux>> items) {
- // do nothing - we just reference this method from the test
- }
-
- static class FirstLevelWrapper {
-
- 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 other = (FirstLevelWrapper) obj;
- return Objects.equals(data, other.data) && id == other.id;
- }
-
- @Override
- public String toString() {
- return "FirstLevelWrapper [id=" + id + ", data=" + data + "]";
- }
-
- }
-
- static class SecondLevelWrapper {
-
- 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 other = (SecondLevelWrapper) obj;
- return Objects.equals(data, other.data) && Objects.equals(name, other.name);
- }
-
- @Override
- public String toString() {
- return "SecondLevelWrapper [name=" + name + ", data=" + data + "]";
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterBasicTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterBasicTests.java
deleted file mode 100644
index 4060b3a5d..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterBasicTests.java
+++ /dev/null
@@ -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 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 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 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 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 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 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> emit() {
- AtomicInteger atomicInteger = new AtomicInteger();
-
- return IntegrationFlows
- .from((Supplier) () -> new GenericMessage<>(
- "Hello World!!" + atomicInteger.getAndIncrement()),
- e -> e.poller(p -> p.fixedDelay(1)))
- .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 emit1() {
- return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
- }
-
- @StreamEmitter
- @Output(TestMultiOutboundChannels.OUTPUT2)
- public Flux 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 emit1() {
- return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
- }
-
- }
-
- static class Bar {
-
- @StreamEmitter
- @Output(TestMultiOutboundChannels.OUTPUT2)
- public Flux emit2() {
- return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello FooBar!!" + l);
- }
-
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterValidationTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterValidationTests.java
deleted file mode 100644
index 035796057..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterValidationTests.java
+++ /dev/null
@@ -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 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 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 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 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";
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerGenericFluxInputOutputArgsWithMessageTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerGenericFluxInputOutputArgsWithMessageTests.java
deleted file mode 100644
index 985f8ea1e..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerGenericFluxInputOutputArgsWithMessageTests.java
+++ /dev/null
@@ -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 result = (Message) 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 {
-
- }
-
- public static class TestGenericStringFluxInputOutputArgsWithMessageImpl2
- extends TestGenericFluxInputOutputArgsWithMessage2 {
-
- }
-
- @EnableBinding(Processor.class)
- @EnableAutoConfiguration
- public static class TestGenericFluxInputOutputArgsWithMessage1 {
-
- @StreamListener
- public void receive(@Input(Processor.INPUT) Flux 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 {
-
- @StreamListener(Processor.INPUT)
- public void receive(Flux input, @Output(Processor.OUTPUT) FluxSender output) {
- output.send(input.map(m -> MessageBuilder
- .withPayload((A) m.toString().toUpperCase()).build()));
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerInterruptionTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerInterruptionTests.java
deleted file mode 100644
index 783e66d80..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerInterruptionTests.java
+++ /dev/null
@@ -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
- * 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 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();
- });
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java
deleted file mode 100644
index a1812dce9..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java
+++ /dev/null
@@ -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 result = (Message) 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 input,
- @Output(Processor.OUTPUT) FluxSender output) {
- output.send(input.map(m -> m.toUpperCase()));
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java
deleted file mode 100644
index b691ac675..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java
+++ /dev/null
@@ -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 result = (Message) 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> input,
- @Output(Processor.OUTPUT) FluxSender output) {
- output.send(input.map(m -> MessageBuilder
- .withPayload(m.getPayload().toString().toUpperCase()).build()));
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java
deleted file mode 100644
index efa5d4482..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java
+++ /dev/null
@@ -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 result = (Message) 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> 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()));
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java
deleted file mode 100644
index ddb5b8716..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java
+++ /dev/null
@@ -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 result = (Message) 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> input,
- @Output(Processor.OUTPUT) FluxSender output) {
- output.send(input.map(m -> m.getPayload().toString().toUpperCase())
- .map(o -> MessageBuilder.withPayload(o).build()));
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodTests.java
deleted file mode 100644
index a0bc0d731..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodTests.java
+++ /dev/null
@@ -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 input,
- @Output(Processor.OUTPUT) FluxSender output) {
- output.send(input.map(m -> m.toUpperCase()));
- }
-
- }
-
- @EnableBinding(Processor.class)
- @EnableAutoConfiguration
- public static class ReactorTestReturn5 {
-
- @StreamListener
- public Flux receive(@Input(Processor.INPUT) Flux input) {
- return input.map(m -> m.toUpperCase());
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java
deleted file mode 100644
index 029ee61df..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java
+++ /dev/null
@@ -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 result = (Message) 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 receive(
- @Input(Processor.INPUT) Flux input) {
- return input.map(m -> m.toUpperCase());
- }
-
- }
-
- @EnableBinding(Processor.class)
- @EnableAutoConfiguration
- public static class ReactorTestReturn2 {
-
- @StreamListener(Processor.INPUT)
- @Output(Processor.OUTPUT)
- public Flux receive(Flux input) {
- return input.map(m -> m.toUpperCase());
- }
-
- }
-
- @EnableBinding(Processor.class)
- @EnableAutoConfiguration
- public static class ReactorTestReturn3 {
-
- @StreamListener(Processor.INPUT)
- @SendTo(Processor.OUTPUT)
- public Flux receive(Flux input) {
- return input.map(m -> m.toUpperCase());
- }
-
- }
-
- @EnableBinding(Processor.class)
- @EnableAutoConfiguration
- public static class ReactorTestReturn4 {
-
- @StreamListener
- @SendTo(Processor.OUTPUT)
- public Flux receive(@Input(Processor.INPUT) Flux input) {
- return input.map(m -> m.toUpperCase());
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java
deleted file mode 100644
index 2fba02f57..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java
+++ /dev/null
@@ -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 result = (Message) 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 receive(
- @Input(Processor.INPUT) Flux 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 receive(Flux 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 receive(Flux 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 receive(
- @Input(Processor.INPUT) Flux input) {
- return input.map(m -> {
- if (!m.equals("fail")) {
- return m.toUpperCase();
- }
- else {
- throw new RuntimeException();
- }
- });
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java
deleted file mode 100644
index 0b5a09c32..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java
+++ /dev/null
@@ -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 result = (Message) 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 receive(
- @Input(Processor.INPUT) Flux> input) {
- return input.map(m -> m.getPayload().toUpperCase());
- }
-
- }
-
- @EnableBinding(Processor.class)
- @EnableAutoConfiguration
- public static class ReactorTestReturnWithMessage2 {
-
- @StreamListener(Processor.INPUT)
- public @Output(Processor.OUTPUT) Flux receive(
- Flux> input) {
- return input.map(m -> m.getPayload().toUpperCase());
- }
-
- }
-
- @EnableBinding(Processor.class)
- @EnableAutoConfiguration
- public static class ReactorTestReturnWithMessage3 {
-
- @StreamListener(Processor.INPUT)
- public @SendTo(Processor.OUTPUT) Flux receive(
- Flux> input) {
- return input.map(m -> m.getPayload().toUpperCase());
- }
-
- }
-
- @EnableBinding(Processor.class)
- @EnableAutoConfiguration
- public static class ReactorTestReturnWithMessage4 {
-
- @StreamListener
- public @SendTo(Processor.OUTPUT) Flux receive(
- @Input(Processor.INPUT) Flux> input) {
- return input.map(m -> m.getPayload().toUpperCase());
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java
deleted file mode 100644
index 8db8cc4db..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java
+++ /dev/null
@@ -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 result = (Message) 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 receive(
- @Input(Processor.INPUT) Flux 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 receive(Flux 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 receive(Flux input) {
- return input.map(m -> new BarPojo(m.getMessage()));
- }
-
- }
-
- @EnableBinding(Processor.class)
- @EnableAutoConfiguration
- public static class ReactorTestReturnWithPojo4 {
-
- @StreamListener
- public @SendTo(Processor.OUTPUT) Flux receive(
- @Input(Processor.INPUT) Flux 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;
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerWildCardFluxInputOutputArgsWithMessageTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerWildCardFluxInputOutputArgsWithMessageTests.java
deleted file mode 100644
index 6f7fc03be..000000000
--- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerWildCardFluxInputOutputArgsWithMessageTests.java
+++ /dev/null
@@ -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 result = (Message) 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()));
- }
-
- }
-
-}
diff --git a/spring-cloud-stream-reactive/src/test/resources/logback.xml b/spring-cloud-stream-reactive/src/test/resources/logback.xml
deleted file mode 100644
index 412f0d7d9..000000000
--- a/spring-cloud-stream-reactive/src/test/resources/logback.xml
+++ /dev/null
@@ -1,10 +0,0 @@
-
-
-
- %d{ISO8601} %5p %t %c{2}:%L - %m%n
-
-
-
-
-
-
diff --git a/spring-cloud-stream-schema/foodorder.avro b/spring-cloud-stream-schema/foodorder.avro
deleted file mode 100644
index ca3f36d4d..000000000
Binary files a/spring-cloud-stream-schema/foodorder.avro and /dev/null differ