Added checkstyle

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * 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.
@@ -33,17 +33,16 @@ import org.springframework.util.Assert;
  * {@link reactor.core.publisher.Flux}.
  *
  * @author Soby Chacko
- *
  * @since 1.3.0
  */
 class DefaultFluxSender implements FluxSender {
 
+	private final Consumer<Object> consumer;
+
 	private Log log = LogFactory.getLog(DefaultFluxSender.class);
 
 	private volatile Disposable disposable;
 
-	private final Consumer<Object> consumer;
-
 	DefaultFluxSender(Consumer<Object> consumer) {
 		Assert.notNull(consumer, "Consumer must not be null");
 		this.consumer = consumer;
@@ -54,12 +53,8 @@ class DefaultFluxSender implements FluxSender {
 		MonoProcessor<Void> sendResult = MonoProcessor.create();
 		// add error handling and reconnect in the event of an error
 		this.disposable = flux
-				.doOnError(e -> this.log.error("Error during processing: ", e))
-				.retry()
-				.subscribe(
-						this.consumer,
-						sendResult::onError,
-						sendResult::onComplete);
+				.doOnError(e -> this.log.error("Error during processing: ", e)).retry()
+				.subscribe(this.consumer, sendResult::onError, sendResult::onComplete);
 		return sendResult;
 	}
 

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

@@ -22,12 +22,11 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 /**
- * Used for {@link org.springframework.cloud.stream.annotation.StreamListener}
- * and {@link org.springframework.cloud.stream.reactive.StreamEmitter} arguments
- * annotated with {@link org.springframework.cloud.stream.annotation.Output}.
+ * Used for {@link org.springframework.cloud.stream.annotation.StreamListener} and
+ * {@link org.springframework.cloud.stream.reactive.StreamEmitter} arguments annotated
+ * with {@link org.springframework.cloud.stream.annotation.Output}.
  *
  * @author Marius Bogoevici
- *
  * @see reactor.core.Disposable
  */
 public interface FluxSender extends Closeable {

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

@@ -42,9 +42,8 @@ public class MessageChannelToFluxSenderParameterAdapter
 
 	@Override
 	public FluxSender adapt(MessageChannel bindingTarget, MethodParameter parameter) {
-		return new DefaultFluxSender(result ->
-				bindingTarget.send(result instanceof Message<?>
-						? (Message<?>) result
+		return new DefaultFluxSender(result -> bindingTarget
+				.send(result instanceof Message<?> ? (Message<?>) result
 						: MessageBuilder.withPayload(result).build()));
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * 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.
@@ -32,6 +32,7 @@ import org.springframework.util.Assert;
 /**
  * Adapts an {@link org.springframework.cloud.stream.annotation.Input} annotated
  * {@link MessageChannel} to a {@link Flux}.
+ *
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
  * @author Vinicius Carvalho
@@ -41,7 +42,8 @@ public class MessageChannelToInputFluxParameterAdapter
 
 	private final CompositeMessageConverter messageConverter;
 
-	public MessageChannelToInputFluxParameterAdapter(CompositeMessageConverter messageConverter) {
+	public MessageChannelToInputFluxParameterAdapter(
+			CompositeMessageConverter messageConverter) {
 		Assert.notNull(messageConverter, "cannot not be null");
 		this.messageConverter = messageConverter;
 	}
@@ -53,19 +55,22 @@ public class MessageChannelToInputFluxParameterAdapter
 	}
 
 	@Override
-	public Flux<?> adapt(final SubscribableChannel bindingTarget, MethodParameter parameter) {
-		final ResolvableType fluxResolvableType = ResolvableType.forMethodParameter(parameter);
+	public Flux<?> adapt(final SubscribableChannel bindingTarget,
+			MethodParameter parameter) {
+		final ResolvableType fluxResolvableType = ResolvableType
+				.forMethodParameter(parameter);
 		final ResolvableType fluxTypeParameter = fluxResolvableType.getGeneric(0);
 		final Class<?> fluxTypeParameterRawClass = fluxTypeParameter.getRawClass();
-		final Class<?> fluxTypeParameterClass = (fluxTypeParameterRawClass != null) ? fluxTypeParameterRawClass
-				: Object.class;
+		final Class<?> fluxTypeParameterClass = (fluxTypeParameterRawClass != null)
+				? fluxTypeParameterRawClass : Object.class;
 
 		final Object monitor = new Object();
 
 		if (Message.class.isAssignableFrom(fluxTypeParameterClass)) {
 
 			final ResolvableType payloadTypeParameter = fluxTypeParameter.getGeneric(0);
-			final Class<?> payloadTypeParameterRawClass = payloadTypeParameter.getRawClass();
+			final Class<?> payloadTypeParameterRawClass = payloadTypeParameter
+					.getRawClass();
 			final Class<?> payloadTypeParameterClass = (payloadTypeParameterRawClass != null)
 					? payloadTypeParameterRawClass : Object.class;
 
@@ -73,12 +78,14 @@ public class MessageChannelToInputFluxParameterAdapter
 				MessageHandler messageHandler = message -> {
 					synchronized (monitor) {
 
-						if (payloadTypeParameterClass.isAssignableFrom(message.getPayload().getClass())) {
+						if (payloadTypeParameterClass
+								.isAssignableFrom(message.getPayload().getClass())) {
 							emitter.next(message);
 						}
 						else {
 							emitter.next(MessageBuilder.createMessage(
-									this.messageConverter.fromMessage(message, payloadTypeParameterClass),
+									this.messageConverter.fromMessage(message,
+											payloadTypeParameterClass),
 									message.getHeaders()));
 						}
 					}
@@ -91,11 +98,13 @@ public class MessageChannelToInputFluxParameterAdapter
 			return Flux.create(emitter -> {
 				MessageHandler messageHandler = message -> {
 					synchronized (monitor) {
-						if (fluxTypeParameterClass.isAssignableFrom(message.getPayload().getClass())) {
+						if (fluxTypeParameterClass
+								.isAssignableFrom(message.getPayload().getClass())) {
 							emitter.next(message.getPayload());
 						}
 						else {
-							emitter.next(this.messageConverter.fromMessage(message, fluxTypeParameterClass));
+							emitter.next(this.messageConverter.fromMessage(message,
+									fluxTypeParameterClass));
 						}
 					}
 				};
@@ -104,4 +113,5 @@ public class MessageChannelToInputFluxParameterAdapter
 			}).publish().autoConnect();
 		}
 	}
+
 }

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

@@ -21,7 +21,6 @@ import java.io.Closeable;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Publisher;
-
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 
@@ -49,15 +48,14 @@ public class PublisherToMessageChannelResultAdapter
 				&& MessageChannel.class.isAssignableFrom(bindingTarget);
 	}
 
-	public Closeable adapt(Publisher<?> streamListenerResult, MessageChannel 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()));
+				.subscribe(result -> bindingTarget
+						.send(result instanceof Message<?> ? (Message<?>) result
+								: MessageBuilder.withPayload(result).build()));
 
 		return disposable::dispose;
 	}

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

@@ -30,6 +30,11 @@ import org.springframework.context.annotation.Configuration;
 @ConditionalOnBean(BindingService.class)
 public class ReactiveSupportAutoConfiguration {
 
+	@Bean
+	public static StreamEmitterAnnotationBeanPostProcessor streamEmitterAnnotationBeanPostProcessor() {
+		return new StreamEmitterAnnotationBeanPostProcessor();
+	}
+
 	@Bean
 	@ConditionalOnMissingBean(MessageChannelToInputFluxParameterAdapter.class)
 	public MessageChannelToInputFluxParameterAdapter messageChannelToInputFluxArgumentAdapter(
@@ -50,8 +55,4 @@ public class ReactiveSupportAutoConfiguration {
 		return new PublisherToMessageChannelResultAdapter();
 	}
 
-	@Bean
-	public static StreamEmitterAnnotationBeanPostProcessor streamEmitterAnnotationBeanPostProcessor() {
-		return new StreamEmitterAnnotationBeanPostProcessor();
-	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * 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.
@@ -30,12 +30,13 @@ 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.
+ * 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.
+ * 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:
  *
@@ -44,14 +45,14 @@ import org.springframework.cloud.stream.annotation.Output;
  * <pre class="code">
  * &#064;StreamEmitter
  * &#064;Output(Source.OUTPUT)
- * public Flux<String> emit() {
+ * public Flux&lt;String&gt; emit() {
  * 	return Flux.intervalMillis(1000)
  * 		.map(l -> "Hello World!!");
  * }
  * </pre>
  *
- * The following examples show how a void return type can be used on a method with StreamEmitter and how the
- * method signatures could be used in a flexible manner.
+ * The following examples show how a void return type can be used on a method with
+ * StreamEmitter and how the method signatures could be used in a flexible manner.
  *
  * <pre class="code">
  * &#064;StreamEmitter
@@ -91,4 +92,6 @@ import org.springframework.cloud.stream.annotation.Output;
 @Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
-public @interface StreamEmitter {}
+public @interface StreamEmitter {
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * 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.
@@ -54,17 +54,22 @@ import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
 
 /**
- * {@link BeanPostProcessor} that handles {@link StreamEmitter} annotations found on bean methods.
+ * {@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 {
+public class StreamEmitterAnnotationBeanPostProcessor implements BeanPostProcessor,
+		SmartInitializingSingleton, ApplicationContextAware, SmartLifecycle {
 
-	private static final Log log = LogFactory.getLog(StreamEmitterAnnotationBeanPostProcessor.class);
+	private static final Log log = LogFactory
+			.getLog(StreamEmitterAnnotationBeanPostProcessor.class);
+
+	private final List<Closeable> closeableFluxResources = new ArrayList<>();
+
+	private final Lock lock = new ReentrantLock();
 
 	@SuppressWarnings("rawtypes")
 	private Collection<StreamListenerParameterAdapter> parameterAdapters;
@@ -72,148 +77,39 @@ public class StreamEmitterAnnotationBeanPostProcessor
 	@SuppressWarnings("rawtypes")
 	private Collection<StreamListenerResultAdapter> resultAdapters;
 
-	private final List<Closeable> closeableFluxResources = new ArrayList<>();
-
 	private ConfigurableApplicationContext applicationContext;
 
 	private MultiValueMap<Object, Method> mappedStreamEmitterMethods = new LinkedMultiValueMap<>();
 
 	private volatile boolean running;
 
-	private final Lock lock = new ReentrantLock();
-
-	@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)) {
-						mappedStreamEmitterMethods.add(bean, method);
-					}
-				}, ReflectionUtils.USER_DECLARED_METHODS);
-		return bean;
-	}
-
-	@Override
-	public void start() {
-		try {
-			lock.lock();
-			if (!running) {
-				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 {
-			lock.unlock();
-		}
-	}
-
-	@SuppressWarnings({ "rawtypes", "unchecked" })
-	private void invokeSetupMethodOnToTargetChannel(Method method, Object bean, String outboundName) {
-		Object[] arguments = new Object[method.getParameterCount()];
-		Object targetBean = null;
-		for (int parameterIndex = 0; parameterIndex < arguments.length; parameterIndex++) {
-			MethodParameter methodParameter = new SynthesizingMethodParameter(method, parameterIndex);
-			Class<?> parameterType = methodParameter.getParameterType();
-			Object targetReferenceValue = null;
-			if (methodParameter.hasParameterAnnotation(Output.class)) {
-				targetReferenceValue = AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Output.class));
-			}
-			else if (arguments.length == 1 && StringUtils.hasText(outboundName)) {
-				targetReferenceValue = outboundName;
-			}
-			if (targetReferenceValue != null) {
-				targetBean = this.applicationContext.getBean((String) targetReferenceValue);
-				for (StreamListenerParameterAdapter<?, Object> streamListenerParameterAdapter : this.parameterAdapters) {
-					if (streamListenerParameterAdapter.supports(targetBean.getClass(), methodParameter)) {
-						arguments[parameterIndex] = streamListenerParameterAdapter.adapt(targetBean,
-								methodParameter);
-						if (arguments[parameterIndex] instanceof FluxSender) {
-							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);
-					closeableFluxResources.add(fluxDisposable);
-					streamListenerResultAdapterFound = true;
-					break;
-				}
-			}
-			Assert.state(streamListenerResultAdapterFound,
-					StreamEmitterErrorMessages.CANNOT_CONVERT_RETURN_TYPE_TO_ANY_AVAILABLE_RESULT_ADAPTERS);
-		}
-	}
-
-	private static void validateStreamEmitterMethod(Method method, int outputAnnotationCount,
-			String methodAnnotatedOutboundName) {
+	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);
+			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);
+			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);
+					MethodParameter methodParameter = new MethodParameter(method,
+							parameterIndex);
 					if (methodParameter.hasParameterAnnotation(Output.class)) {
-						String outboundName = (String) AnnotationUtils
-								.getValue(methodParameter.getParameterAnnotation(Output.class));
+						String outboundName = (String) AnnotationUtils.getValue(
+								methodParameter.getParameterAnnotation(Output.class));
 						Assert.isTrue(StringUtils.hasText(outboundName),
 								StreamEmitterErrorMessages.INVALID_OUTBOUND_NAME);
 					}
@@ -226,6 +122,131 @@ public class StreamEmitterAnnotationBeanPostProcessor
 		}
 	}
 
+	@Override
+	public final void setApplicationContext(ApplicationContext applicationContext)
+			throws BeansException {
+		Assert.isTrue(applicationContext instanceof ConfigurableApplicationContext,
+				"ConfigurableApplicationContext is required");
+		this.applicationContext = (ConfigurableApplicationContext) applicationContext;
+	}
+
+	@Override
+	public void afterSingletonsInstantiated() {
+		this.parameterAdapters = this.applicationContext
+				.getBeansOfType(StreamListenerParameterAdapter.class).values();
+		this.resultAdapters = new ArrayList<>(this.applicationContext
+				.getBeansOfType(StreamListenerResultAdapter.class).values());
+		this.resultAdapters.add(new MessageChannelStreamListenerResultAdapter());
+	}
+
+	@Override
+	public Object postProcessAfterInitialization(final Object bean, final String beanName)
+			throws BeansException {
+		Class<?> targetClass = AopUtils.getTargetClass(bean);
+		ReflectionUtils.doWithMethods(targetClass, method -> {
+			if (AnnotatedElementUtils.isAnnotated(method, StreamEmitter.class)) {
+				this.mappedStreamEmitterMethods.add(bean, method);
+			}
+		}, ReflectionUtils.USER_DECLARED_METHODS);
+		return bean;
+	}
+
+	@Override
+	public void start() {
+		try {
+			this.lock.lock();
+			if (!this.running) {
+				this.mappedStreamEmitterMethods.forEach((k, v) -> v.forEach(item -> {
+					Assert.isTrue(item.getAnnotation(Input.class) == null,
+							StreamEmitterErrorMessages.INPUT_ANNOTATIONS_ARE_NOT_ALLOWED);
+					String methodAnnotatedOutboundName = StreamAnnotationCommonMethodUtils
+							.getOutboundBindingTargetName(item);
+					int outputAnnotationCount = StreamAnnotationCommonMethodUtils
+							.outputAnnotationCount(item);
+					validateStreamEmitterMethod(item, outputAnnotationCount,
+							methodAnnotatedOutboundName);
+					invokeSetupMethodOnToTargetChannel(item, k,
+							methodAnnotatedOutboundName);
+				}));
+				this.running = true;
+			}
+		}
+		finally {
+			this.lock.unlock();
+		}
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private void invokeSetupMethodOnToTargetChannel(Method method, Object bean,
+			String outboundName) {
+		Object[] arguments = new Object[method.getParameterCount()];
+		Object targetBean = null;
+		for (int parameterIndex = 0; parameterIndex < arguments.length; parameterIndex++) {
+			MethodParameter methodParameter = new SynthesizingMethodParameter(method,
+					parameterIndex);
+			Class<?> parameterType = methodParameter.getParameterType();
+			Object targetReferenceValue = null;
+			if (methodParameter.hasParameterAnnotation(Output.class)) {
+				targetReferenceValue = AnnotationUtils
+						.getValue(methodParameter.getParameterAnnotation(Output.class));
+			}
+			else if (arguments.length == 1 && StringUtils.hasText(outboundName)) {
+				targetReferenceValue = outboundName;
+			}
+			if (targetReferenceValue != null) {
+				targetBean = this.applicationContext
+						.getBean((String) targetReferenceValue);
+				for (StreamListenerParameterAdapter<?, Object> streamListenerParameterAdapter : this.parameterAdapters) {
+					if (streamListenerParameterAdapter.supports(targetBean.getClass(),
+							methodParameter)) {
+						arguments[parameterIndex] = streamListenerParameterAdapter
+								.adapt(targetBean, methodParameter);
+						if (arguments[parameterIndex] instanceof FluxSender) {
+							this.closeableFluxResources
+									.add((FluxSender) arguments[parameterIndex]);
+						}
+						break;
+					}
+				}
+				Assert.notNull(arguments[parameterIndex],
+						"Cannot convert argument " + parameterIndex + " of " + method
+								+ "from " + targetBean.getClass() + " to "
+								+ parameterType);
+			}
+			else {
+				throw new IllegalStateException(
+						StreamEmitterErrorMessages.ATLEAST_ONE_OUTPUT);
+			}
+		}
+		Object result;
+		try {
+			result = method.invoke(bean, arguments);
+		}
+		catch (Exception e) {
+			throw new BeanInitializationException(
+					"Cannot setup StreamEmitter for " + method, e);
+		}
+
+		if (!Void.TYPE.equals(method.getReturnType())) {
+			if (targetBean == null) {
+				targetBean = this.applicationContext.getBean(outboundName);
+			}
+			boolean streamListenerResultAdapterFound = false;
+			for (StreamListenerResultAdapter streamListenerResultAdapter : this.resultAdapters) {
+				if (streamListenerResultAdapter.supports(result.getClass(),
+						targetBean.getClass())) {
+					Closeable fluxDisposable = streamListenerResultAdapter.adapt(result,
+							targetBean);
+					this.closeableFluxResources.add(fluxDisposable);
+					streamListenerResultAdapterFound = true;
+					break;
+				}
+			}
+			Assert.state(streamListenerResultAdapterFound,
+					StreamEmitterErrorMessages.CANNOT_CONVERT_RETURN_TYPE_TO_ANY_AVAILABLE_RESULT_ADAPTERS);
+		}
+	}
+
 	@Override
 	public boolean isAutoStartup() {
 		return true;
@@ -244,7 +265,7 @@ public class StreamEmitterAnnotationBeanPostProcessor
 		try {
 			this.lock.lock();
 			if (this.running) {
-				for (Closeable closeable : closeableFluxResources) {
+				for (Closeable closeable : this.closeableFluxResources) {
 					try {
 						closeable.close();
 					}

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

@@ -24,27 +24,24 @@ import org.springframework.cloud.stream.binding.StreamAnnotationErrorMessages;
  */
 abstract class StreamEmitterErrorMessages extends StreamAnnotationErrorMessages {
 
-	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 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";
 
-	static final String CANNOT_CONVERT_RETURN_TYPE_TO_ANY_AVAILABLE_RESULT_ADAPTERS =
-			"No suitable adapters are found that can convert the return type";
-
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * 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.
@@ -46,7 +46,8 @@ public class MessageChannelToInputFluxParameterAdapterTests {
 	public void testWrapperFluxSupportsMultipleSubscriptions() throws Exception {
 		List<String> results = Collections.synchronizedList(new ArrayList<>());
 		CountDownLatch latch = new CountDownLatch(4);
-		final MessageChannelToInputFluxParameterAdapter messageChannelToInputFluxParameterAdapter = new MessageChannelToInputFluxParameterAdapter(
+		final MessageChannelToInputFluxParameterAdapter messageChannelToInputFluxParameterAdapter;
+		messageChannelToInputFluxParameterAdapter = new MessageChannelToInputFluxParameterAdapter(
 				new CompositeMessageConverter(
 						Collections.singleton(new MappingJackson2MessageConverter())));
 		final Method processMethod = ReflectionUtils.findMethod(
@@ -79,4 +80,5 @@ public class MessageChannelToInputFluxParameterAdapterTests {
 	public void process(Flux<Message<?>> message) {
 		// do nothing - we just reference this method from the test
 	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * 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.
@@ -16,7 +16,6 @@
 
 package org.springframework.cloud.stream.reactive;
 
-
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
@@ -51,10 +50,87 @@ import static org.assertj.core.api.Assertions.assertThat;
  */
 public class StreamEmitterBasicTests {
 
+	private static void receiveAndValidate(ConfigurableApplicationContext context)
+			throws InterruptedException {
+		Source source = context.getBean(Source.class);
+		MessageCollector messageCollector = context.getBean(MessageCollector.class);
+		List<String> messages = new ArrayList<>();
+		for (int i = 0; i < 1000; i++) {
+			messages.add((String) messageCollector.forChannel(source.output())
+					.poll(5000, TimeUnit.MILLISECONDS).getPayload());
+		}
+		for (int i = 0; i < 1000; i++) {
+			assertThat(new String(messages.get(i))).isEqualTo("HELLO WORLD!!" + i);
+		}
+	}
+
+	private static void receiveAndValidateMultipleOutputs(
+			ConfigurableApplicationContext context) throws InterruptedException {
+		TestMultiOutboundChannels source = context
+				.getBean(TestMultiOutboundChannels.class);
+		MessageCollector messageCollector = context.getBean(MessageCollector.class);
+		List<String> messages = new ArrayList<>();
+		assertMessages(source.output1(), messageCollector, messages);
+		messages.clear();
+		assertMessages(source.output2(), messageCollector, messages);
+		messages.clear();
+		assertMessages(source.output3(), messageCollector, messages);
+		messages.clear();
+	}
+
+	private static void receiveAndValidateMultiStreamEmittersInSameContext(
+			ConfigurableApplicationContext context1) throws InterruptedException {
+		TestMultiOutboundChannels source1 = context1
+				.getBean(TestMultiOutboundChannels.class);
+		MessageCollector messageCollector = context1.getBean(MessageCollector.class);
+
+		List<String> messages = new ArrayList<>();
+		assertMessagesX(source1.output1(), messageCollector, messages);
+		messages.clear();
+		assertMessagesY(source1.output2(), messageCollector, messages);
+		messages.clear();
+	}
+
+	private static void assertMessages(MessageChannel channel,
+			MessageCollector messageCollector, List<String> messages)
+			throws InterruptedException {
+		for (int i = 0; i < 1000; i++) {
+			messages.add((String) messageCollector.forChannel(channel)
+					.poll(5000, TimeUnit.MILLISECONDS).getPayload());
+		}
+		for (int i = 0; i < 1000; i++) {
+			assertThat(new String(messages.get(i))).isEqualTo("Hello World!!" + i);
+		}
+	}
+
+	private static void assertMessagesX(MessageChannel channel,
+			MessageCollector messageCollector, List<String> messages)
+			throws InterruptedException {
+		for (int i = 0; i < 1000; i++) {
+			messages.add((String) messageCollector.forChannel(channel)
+					.poll(5000, TimeUnit.MILLISECONDS).getPayload());
+		}
+		for (int i = 0; i < 1000; i++) {
+			assertThat(new String(messages.get(i))).isEqualTo("Hello World!!" + i);
+		}
+	}
+
+	private static void assertMessagesY(MessageChannel channel,
+			MessageCollector messageCollector, List<String> messages)
+			throws InterruptedException {
+		for (int i = 0; i < 1000; i++) {
+			messages.add((String) messageCollector.forChannel(channel)
+					.poll(5000, TimeUnit.MILLISECONDS).getPayload());
+		}
+		for (int i = 0; i < 1000; i++) {
+			assertThat(new String(messages.get(i))).isEqualTo("Hello FooBar!!" + i);
+		}
+	}
+
 	@Test
 	public void testFluxReturnAndOutputMethodLevel() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication.run(TestFluxReturnAndOutputMethodLevel.class,
-				"--server.port=0",
+		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");
@@ -64,8 +140,8 @@ public class StreamEmitterBasicTests {
 
 	@Test
 	public void testVoidReturnAndOutputMethodParameter() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication.run(TestVoidReturnAndOutputMethodParameter.class,
-				"--server.port=0",
+		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");
@@ -75,8 +151,8 @@ public class StreamEmitterBasicTests {
 
 	@Test
 	public void testVoidReturnAndOutputAtMethodLevel() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication.run(TestVoidReturnAndOutputAtMethodLevel.class,
-				"--server.port=0",
+		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");
@@ -86,8 +162,8 @@ public class StreamEmitterBasicTests {
 
 	@Test
 	public void testVoidReturnAndMultipleOutputMethodParameters() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication.run(TestVoidReturnAndMultipleOutputMethodParameters.class,
-				"--server.port=0",
+		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",
@@ -100,8 +176,8 @@ public class StreamEmitterBasicTests {
 
 	@Test
 	public void testMultipleStreamEmitterMethods() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication.run(TestMultipleStreamEmitterMethods.class,
-				"--server.port=0",
+		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",
@@ -114,8 +190,8 @@ public class StreamEmitterBasicTests {
 
 	@Test
 	public void testSameAppContextWithMultipleStreamEmitters() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication.run(TestSameAppContextWithMultipleStreamEmitters.class,
-				"--server.port=0",
+		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",
@@ -126,66 +202,23 @@ public class StreamEmitterBasicTests {
 		context.close();
 	}
 
-	private static void receiveAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
-		Source source = context.getBean(Source.class);
-		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		List<String> messages = new ArrayList<>();
-		for (int i = 0; i < 1000; i++) {
-			messages.add((String) messageCollector.forChannel(source.output()).poll(5000, TimeUnit.MILLISECONDS).getPayload());
-		}
-		for (int i = 0; i < 1000; i++) {
-			assertThat(new String(messages.get(i))).isEqualTo("HELLO WORLD!!" + i);
-		}
-	}
+	interface TestMultiOutboundChannels {
 
-	private static void receiveAndValidateMultipleOutputs(ConfigurableApplicationContext context) throws InterruptedException {
-		TestMultiOutboundChannels source = context.getBean(TestMultiOutboundChannels.class);
-		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		List<String> messages = new ArrayList<>();
-		assertMessages(source.output1(), messageCollector, messages);
-		messages.clear();
-		assertMessages(source.output2(), messageCollector, messages);
-		messages.clear();
-		assertMessages(source.output3(), messageCollector, messages);
-		messages.clear();
-	}
+		String OUTPUT1 = "output1";
 
-	private static void receiveAndValidateMultiStreamEmittersInSameContext(ConfigurableApplicationContext context1) throws InterruptedException {
-		TestMultiOutboundChannels source1 = context1.getBean(TestMultiOutboundChannels.class);
-		MessageCollector messageCollector = context1.getBean(MessageCollector.class);
+		String OUTPUT2 = "output2";
 
-		List<String> messages = new ArrayList<>();
-		assertMessagesX(source1.output1(), messageCollector, messages);
-		messages.clear();
-		assertMessagesY(source1.output2(), messageCollector, messages);
-		messages.clear();
-	}
+		String OUTPUT3 = "output3";
 
-	private static void assertMessages(MessageChannel channel, MessageCollector messageCollector, List<String> messages) throws InterruptedException {
-		for (int i = 0; i < 1000; i++) {
-			messages.add((String) messageCollector.forChannel(channel).poll(5000, TimeUnit.MILLISECONDS).getPayload());
-		}
-		for (int i = 0; i < 1000; i++) {
-			assertThat(new String(messages.get(i))).isEqualTo("Hello World!!" + i);
-		}
-	}
+		@Output(TestMultiOutboundChannels.OUTPUT1)
+		MessageChannel output1();
 
-	private static void assertMessagesX(MessageChannel channel, MessageCollector messageCollector, List<String> messages) throws InterruptedException {
-		for (int i = 0; i < 1000; i++) {
-			messages.add((String) messageCollector.forChannel(channel).poll(5000, TimeUnit.MILLISECONDS).getPayload());
-		}
-		for (int i = 0; i < 1000; i++) {
-			assertThat(new String(messages.get(i))).isEqualTo("Hello World!!" + i);
-		}
-	}
+		@Output(TestMultiOutboundChannels.OUTPUT2)
+		MessageChannel output2();
+
+		@Output(TestMultiOutboundChannels.OUTPUT3)
+		MessageChannel output3();
 
-	private static void assertMessagesY(MessageChannel channel, MessageCollector messageCollector, List<String> messages) throws InterruptedException {
-		for (int i = 0; i < 1000; i++) {
-			messages.add((String) messageCollector.forChannel(channel).poll(5000, TimeUnit.MILLISECONDS).getPayload());
-		}
-		for (int i = 0; i < 1000; i++) {
-			assertThat(new String(messages.get(i))).isEqualTo("Hello FooBar!!" + i);
-		}
 	}
 
 	@EnableBinding(Processor.class)
@@ -197,11 +230,11 @@ public class StreamEmitterBasicTests {
 		@Bean
 		public Publisher<Message<String>> emit() {
 			AtomicInteger atomicInteger = new AtomicInteger();
-			return IntegrationFlows.from(() ->
-							new GenericMessage<>("Hello World!!" + atomicInteger.getAndIncrement()),
-					e -> e.poller(p -> p.fixedDelay(1)))
-					.<String, String>transform(String::toUpperCase)
-					.toReactivePublisher();
+			return IntegrationFlows
+					.from(() -> new GenericMessage<>(
+							"Hello World!!" + atomicInteger.getAndIncrement()),
+							e -> e.poller(p -> p.fixedDelay(1)))
+					.<String, String>transform(String::toUpperCase).toReactivePublisher();
 		}
 
 	}
@@ -212,10 +245,10 @@ public class StreamEmitterBasicTests {
 
 		@StreamEmitter
 		public void emit(@Output(Source.OUTPUT) FluxSender output) {
-			output.send(Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l)
+			output.send(Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l)
 					.map(String::toUpperCase));
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -225,10 +258,10 @@ public class StreamEmitterBasicTests {
 		@StreamEmitter
 		@Output(Source.OUTPUT)
 		public void emit(FluxSender output) {
-			output.send(Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l)
+			output.send(Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l)
 					.map(String::toUpperCase));
 		}
+
 	}
 
 	@EnableBinding(TestMultiOutboundChannels.class)
@@ -239,13 +272,14 @@ public class StreamEmitterBasicTests {
 		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));
+			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)
@@ -255,22 +289,21 @@ public class StreamEmitterBasicTests {
 		@StreamEmitter
 		@Output(TestMultiOutboundChannels.OUTPUT1)
 		public Flux<String> emit1() {
-			return Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l);
+			return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
 		}
 
 		@StreamEmitter
 		@Output(TestMultiOutboundChannels.OUTPUT2)
 		public Flux<String> emit2() {
-			return Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l);
+			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));
+			outputX.send(
+					Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
 		}
+
 	}
 
 	@EnableBinding(TestMultiOutboundChannels.class)
@@ -292,9 +325,9 @@ public class StreamEmitterBasicTests {
 			@StreamEmitter
 			@Output(TestMultiOutboundChannels.OUTPUT1)
 			public Flux<String> emit1() {
-				return Flux.interval(Duration.ofMillis(1))
-						.map(l -> "Hello World!!" + l);
+				return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
 			}
+
 		}
 
 		static class Bar {
@@ -302,28 +335,11 @@ public class StreamEmitterBasicTests {
 			@StreamEmitter
 			@Output(TestMultiOutboundChannels.OUTPUT2)
 			public Flux<String> emit2() {
-				return Flux.interval(Duration.ofMillis(1))
-						.map(l -> "Hello FooBar!!" + l);
+				return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello FooBar!!" + l);
 			}
+
 		}
-	}
-
-	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();
 
 	}
+
 }

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

@@ -121,13 +121,16 @@ public class StreamEmitterValidationTests {
 	public void testVoidReturnTypeMultipleMethodParametersWithOneMissingOutput() {
 		try {
 			AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
-			context.register(TestVoidReturnTypeMultipleMethodParametersWithOneMissingOutput.class);
+			context.register(
+					TestVoidReturnTypeMultipleMethodParametersWithOneMissingOutput.class);
 			context.refresh();
 			context.close();
-			fail("Expected exception: " + OUTPUT_ANNOTATION_MISSING_ON_METHOD_PARAMETERS_VOID_RETURN_TYPE);
+			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);
+			assertThat(e.getMessage()).contains(
+					OUTPUT_ANNOTATION_MISSING_ON_METHOD_PARAMETERS_VOID_RETURN_TYPE);
 		}
 	}
 
@@ -180,10 +183,12 @@ public class StreamEmitterValidationTests {
 			context.register(TestReturnTypeNotSupported.class);
 			context.refresh();
 			context.close();
-			fail("Expected exception: " + CANNOT_CONVERT_RETURN_TYPE_TO_ANY_AVAILABLE_RESULT_ADAPTERS);
+			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);
+			assertThat(e.getMessage()).contains(
+					CANNOT_CONVERT_RETURN_TYPE_TO_ANY_AVAILABLE_RESULT_ADAPTERS);
 		}
 	}
 
@@ -194,9 +199,10 @@ public class StreamEmitterValidationTests {
 		@StreamEmitter
 		@Output(Source.OUTPUT)
 		public void receive(@Output(Source.OUTPUT) FluxSender output) {
-			output.send(Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l));
+			output.send(
+					Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -205,9 +211,9 @@ public class StreamEmitterValidationTests {
 
 		@StreamEmitter
 		public Flux<String> emit() {
-			return Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l);
+			return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -216,9 +222,10 @@ public class StreamEmitterValidationTests {
 
 		@StreamEmitter
 		public void emit(FluxSender output) {
-			output.send(Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l));
+			output.send(
+					Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -227,9 +234,9 @@ public class StreamEmitterValidationTests {
 
 		@StreamEmitter
 		public Flux<String> emit(@Output(Source.OUTPUT) FluxSender output) {
-			return Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l);
+			return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -239,9 +246,9 @@ public class StreamEmitterValidationTests {
 		@StreamEmitter
 		@Output(Source.OUTPUT)
 		public Flux<String> receive(FluxSender output) {
-			return Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l);
+			return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
 		}
+
 	}
 
 	@EnableBinding(StreamEmitterBasicTests.TestMultiOutboundChannels.class)
@@ -249,16 +256,18 @@ public class StreamEmitterValidationTests {
 	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));
+		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)
@@ -268,9 +277,10 @@ public class StreamEmitterValidationTests {
 		@StreamEmitter
 		@Output("")
 		public void receive(FluxSender output) {
-			output.send(Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l));
+			output.send(
+					Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
 		}
+
 	}
 
 	@EnableBinding(StreamEmitterBasicTests.TestMultiOutboundChannels.class)
@@ -279,9 +289,10 @@ public class StreamEmitterValidationTests {
 
 		@StreamEmitter
 		public void emit(@Output("") FluxSender output1) {
-			output1.send(Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l));
+			output1.send(
+					Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l));
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -292,9 +303,9 @@ public class StreamEmitterValidationTests {
 		@Output(Source.OUTPUT)
 		@Input(Processor.INPUT)
 		public Flux<String> emit() {
-			return Flux.interval(Duration.ofMillis(1))
-					.map(l -> "Hello World!!" + l);
+			return Flux.interval(Duration.ofMillis(1)).map(l -> "Hello World!!" + l);
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -306,5 +317,7 @@ public class StreamEmitterValidationTests {
 		public String emit() {
 			return "hello";
 		}
+
 	}
+
 }

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

@@ -53,8 +53,8 @@ public class StreamListenerGenericFluxInputOutputArgsWithMessageTests {
 		processor.input().send(MessageBuilder.withPayload(sentPayload)
 				.setHeader("contentType", "text/plain").build());
 		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		Message<String> result = (Message<String>) messageCollector.forChannel(processor.output()).poll(1000,
-				TimeUnit.MILLISECONDS);
+		Message<String> result = (Message<String>) messageCollector
+				.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
 		assertThat(result).isNotNull();
 		assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
 	}
@@ -88,10 +88,12 @@ public class StreamListenerGenericFluxInputOutputArgsWithMessageTests {
 
 	public static class TestGenericStringFluxInputOutputArgsWithMessageImpl1
 			extends TestGenericFluxInputOutputArgsWithMessage1<String> {
+
 	}
 
 	public static class TestGenericStringFluxInputOutputArgsWithMessageImpl2
 			extends TestGenericFluxInputOutputArgsWithMessage2<String> {
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -104,6 +106,7 @@ public class StreamListenerGenericFluxInputOutputArgsWithMessageTests {
 			output.send(input.map(m -> MessageBuilder
 					.withPayload((A) m.toString().toUpperCase()).build()));
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -115,5 +118,7 @@ public class StreamListenerGenericFluxInputOutputArgsWithMessageTests {
 			output.send(input.map(m -> MessageBuilder
 					.withPayload((A) m.toString().toUpperCase()).build()));
 		}
+
 	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * 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.
@@ -44,8 +44,8 @@ public class StreamListenerInterruptionTests {
 
 	@Test
 	public void testSubscribersNotInterrupted() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication.run(TestTimeWindows.class,
-				"--server.port=0");
+		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());
@@ -68,11 +68,12 @@ public class StreamListenerInterruptionTests {
 		@StreamListener
 		public void receive(@Input(Sink.INPUT) Flux<String> input) {
 			input.window(Duration.ofMillis(500), Duration.ofMillis(100))
-					.flatMap(w -> w.reduce("", (x, y) -> x + y))
-					.subscribe(x -> {
-						interruptionState = Thread.currentThread().isInterrupted();
-						latch.countDown();
+					.flatMap(w -> w.reduce("", (x, y) -> x + y)).subscribe(x -> {
+						this.interruptionState = Thread.currentThread().isInterrupted();
+						this.latch.countDown();
 					});
 		}
+
 	}
+
 }

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

@@ -60,20 +60,23 @@ public class StreamListenerReactiveInputOutputArgsTests {
 	}
 
 	@SuppressWarnings("unchecked")
-	private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
+	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());
+		processor.input().send(MessageBuilder.withPayload(sentPayload)
+				.setHeader("contentType", "text/plain").build());
 		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		Message<String> result = (Message<String>) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		Message<String> result = (Message<String>) messageCollector
+				.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
 		assertThat(result).isNotNull();
 		assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
 	}
 
 	@Test
 	public void testInputOutputArgs() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0",
-				"--spring.jmx.enabled=false",
+		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);
@@ -85,8 +88,11 @@ public class StreamListenerReactiveInputOutputArgsTests {
 	public static class ReactorTestInputOutputArgs {
 
 		@StreamListener
-		public void receive(@Input(Processor.INPUT) Flux<String> input, @Output(Processor.OUTPUT) FluxSender output) {
+		public void receive(@Input(Processor.INPUT) Flux<String> input,
+				@Output(Processor.OUTPUT) FluxSender output) {
 			output.send(input.map(m -> m.toUpperCase()));
 		}
+
 	}
+
 }

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

@@ -38,7 +38,6 @@ import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.MessageBuilder;
 
-
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**
@@ -61,20 +60,23 @@ public class StreamListenerReactiveInputOutputArgsWithMessageTests {
 	}
 
 	@SuppressWarnings("unchecked")
-	private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
+	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());
+		processor.input().send(MessageBuilder.withPayload(sentPayload)
+				.setHeader("contentType", "text/plain").build());
 		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		Message<String> result = (Message<String>) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		Message<String> result = (Message<String>) messageCollector
+				.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
 		assertThat(result).isNotNull();
 		assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
 	}
 
 	@Test
 	public void testInputOutputArgs() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0",
-				"--spring.jmx.enabled=false",
+		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);
@@ -91,5 +93,7 @@ public class StreamListenerReactiveInputOutputArgsWithMessageTests {
 			output.send(input.map(m -> MessageBuilder
 					.withPayload(m.getPayload().toString().toUpperCase()).build()));
 		}
+
 	}
+
 }

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

@@ -38,7 +38,6 @@ import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.MessageBuilder;
 
-
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**
@@ -51,35 +50,42 @@ public class StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests {
 
 	private Class<?> configClass;
 
-	public StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests(Class<?> configClass) {
+	public StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests(
+			Class<?> configClass) {
 		this.configClass = configClass;
 	}
 
 	@Parameterized.Parameters
 	public static Collection<?> InputConfigs() {
-		return Collections.singletonList(TestInputOutputArgsWithFluxSenderAndFailure.class);
+		return Collections
+				.singletonList(TestInputOutputArgsWithFluxSenderAndFailure.class);
 	}
 
 	@SuppressWarnings("unchecked")
-	private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
+	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());
+		processor.input().send(MessageBuilder.withPayload(sentPayload)
+				.setHeader("contentType", "text/plain").build());
 		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		Message<String> result = (Message<String>) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		Message<String> result = (Message<String>) messageCollector
+				.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
 		assertThat(result).isNotNull();
 		assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
 	}
 
-	private static void sendFailingMessage(ConfigurableApplicationContext context) throws InterruptedException {
+	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());
+		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",
+		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);
@@ -91,20 +97,20 @@ public class StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests {
 	@EnableBinding(Processor.class)
 	@EnableAutoConfiguration
 	public static class TestInputOutputArgsWithFluxSenderAndFailure {
+
 		@StreamListener
 		public void receive(@Input(Processor.INPUT) Flux<Message<String>> input,
 				@Output(Processor.OUTPUT) FluxSender output) {
-			output.send(input
-					.map(m -> m.getPayload().toString())
-					.map(m -> {
-						if (!m.equals("fail")) {
-							return m.toUpperCase();
-						}
-						else {
-							throw new RuntimeException();
-						}
-					})
-					.map(o -> MessageBuilder.withPayload(o).build()));
+			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()));
 		}
+
 	}
+
 }

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

@@ -60,12 +60,15 @@ public class StreamListenerReactiveInputOutputArgsWithSenderTests {
 	}
 
 	@SuppressWarnings("unchecked")
-	private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
+	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());
+		processor.input().send(MessageBuilder.withPayload(sentPayload)
+				.setHeader("contentType", "text/plain").build());
 		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		Message<String> result = (Message<String>) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		Message<String> result = (Message<String>) messageCollector
+				.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
 		assertThat(result).isNotNull();
 		assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
 	}
@@ -73,8 +76,7 @@ public class StreamListenerReactiveInputOutputArgsWithSenderTests {
 	@Test
 	public void testInputOutputArgsWithFluxSender() throws Exception {
 		ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
-				"--server.port=0",
-				"--spring.jmx.enabled=false",
+				"--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
@@ -87,12 +89,14 @@ public class StreamListenerReactiveInputOutputArgsWithSenderTests {
 	@EnableBinding(Processor.class)
 	@EnableAutoConfiguration
 	public static class ReactorTestInputOutputArgsWithFluxSender {
+
 		@StreamListener
 		public void receive(@Input(Processor.INPUT) Flux<Message<String>> input,
 				@Output(Processor.OUTPUT) FluxSender output) {
-			output.send(input
-					.map(m -> m.getPayload().toString().toUpperCase())
+			output.send(input.map(m -> m.getPayload().toString().toUpperCase())
 					.map(o -> MessageBuilder.withPayload(o).build()));
 		}
+
 	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * 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.
@@ -44,7 +44,8 @@ public class StreamListenerReactiveMethodTests {
 			fail("IllegalArgumentException should have been thrown");
 		}
 		catch (Exception e) {
-			assertThat(e.getMessage()).contains(INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
+			assertThat(e.getMessage())
+					.contains(INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
 		}
 	}
 
@@ -67,9 +68,11 @@ public class StreamListenerReactiveMethodTests {
 	public static class ReactorTestInputOutputArgs {
 
 		@StreamListener(Processor.INPUT)
-		public void receive(Flux<String> input, @Output(Processor.OUTPUT) FluxSender output) {
+		public void receive(Flux<String> input,
+				@Output(Processor.OUTPUT) FluxSender output) {
 			output.send(input.map(m -> m.toUpperCase()));
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -80,5 +83,7 @@ public class StreamListenerReactiveMethodTests {
 		public Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
 			return input.map(m -> m.toUpperCase());
 		}
+
 	}
+
 }

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

@@ -57,25 +57,28 @@ public class StreamListenerReactiveMethodWithReturnTypeTests {
 
 	@Parameterized.Parameters
 	public static Collection<?> InputConfigs() {
-		return Arrays.asList(ReactorTestReturn1.class, ReactorTestReturn2.class, ReactorTestReturn3.class,
-				ReactorTestReturn4.class);
+		return Arrays.asList(ReactorTestReturn1.class, ReactorTestReturn2.class,
+				ReactorTestReturn3.class, ReactorTestReturn4.class);
 	}
 
 	@SuppressWarnings("unchecked")
-	private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
+	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());
+		processor.input().send(MessageBuilder.withPayload(sentPayload)
+				.setHeader("contentType", "text/plain").build());
 		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		Message<String> result = (Message<String>) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		Message<String> result = (Message<String>) messageCollector
+				.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
 		assertThat(result).isNotNull();
 		assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
 	}
 
 	@Test
 	public void testReturn() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0",
-				"--spring.jmx.enabled=false",
+		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);
@@ -89,9 +92,11 @@ public class StreamListenerReactiveMethodWithReturnTypeTests {
 	public static class ReactorTestReturn1 {
 
 		@StreamListener
-		public @Output(Processor.OUTPUT) Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
+		public @Output(Processor.OUTPUT) Flux<String> receive(
+				@Input(Processor.INPUT) Flux<String> input) {
 			return input.map(m -> m.toUpperCase());
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -103,6 +108,7 @@ public class StreamListenerReactiveMethodWithReturnTypeTests {
 		public Flux<String> receive(Flux<String> input) {
 			return input.map(m -> m.toUpperCase());
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -114,6 +120,7 @@ public class StreamListenerReactiveMethodWithReturnTypeTests {
 		public Flux<String> receive(Flux<String> input) {
 			return input.map(m -> m.toUpperCase());
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -125,5 +132,7 @@ public class StreamListenerReactiveMethodWithReturnTypeTests {
 		public Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
 			return input.map(m -> m.toUpperCase());
 		}
+
 	}
+
 }

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

@@ -57,30 +57,35 @@ public class StreamListenerReactiveReturnWithFailureTests {
 
 	@Parameterized.Parameters
 	public static Collection<?> InputConfigs() {
-		return Arrays.asList(ReactorTestReturnWithFailure1.class, ReactorTestReturnWithFailure2.class,
-				ReactorTestReturnWithFailure3.class, ReactorTestReturnWithFailure4.class);
+		return Arrays.asList(ReactorTestReturnWithFailure1.class,
+				ReactorTestReturnWithFailure2.class, ReactorTestReturnWithFailure3.class,
+				ReactorTestReturnWithFailure4.class);
 	}
 
 	@SuppressWarnings("unchecked")
-	private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
+	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());
+		processor.input().send(MessageBuilder.withPayload(sentPayload)
+				.setHeader("contentType", "text/plain").build());
 		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		Message<String> result = (Message<String>) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		Message<String> result = (Message<String>) messageCollector
+				.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
 		assertThat(result).isNotNull();
 		assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
 	}
 
 	private static void sendFailingMessage(ConfigurableApplicationContext context) {
 		Processor processor = context.getBean(Processor.class);
-		processor.input().send(MessageBuilder.withPayload("fail").setHeader("contentType", "text/plain").build());
+		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",
+		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);
@@ -96,16 +101,18 @@ public class StreamListenerReactiveReturnWithFailureTests {
 	public static class ReactorTestReturnWithFailure1 {
 
 		@StreamListener
-		public @Output(Processor.OUTPUT)
-		Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
+		public @Output(Processor.OUTPUT) Flux<String> receive(
+				@Input(Processor.INPUT) Flux<String> input) {
 			return input.map(m -> {
 				if (!m.equals("fail")) {
 					return m.toUpperCase();
-				} else {
+				}
+				else {
 					throw new RuntimeException();
 				}
 			});
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -113,16 +120,17 @@ public class StreamListenerReactiveReturnWithFailureTests {
 	public static class ReactorTestReturnWithFailure2 {
 
 		@StreamListener(Processor.INPUT)
-		public @Output(Processor.OUTPUT)
-		Flux<String> receive(Flux<String> input) {
+		public @Output(Processor.OUTPUT) Flux<String> receive(Flux<String> input) {
 			return input.map(m -> {
 				if (!m.equals("fail")) {
 					return m.toUpperCase();
-				} else {
+				}
+				else {
 					throw new RuntimeException();
 				}
 			});
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -130,16 +138,17 @@ public class StreamListenerReactiveReturnWithFailureTests {
 	public static class ReactorTestReturnWithFailure3 {
 
 		@StreamListener(Processor.INPUT)
-		public @SendTo(Processor.OUTPUT)
-		Flux<String> receive(Flux<String> input) {
+		public @SendTo(Processor.OUTPUT) Flux<String> receive(Flux<String> input) {
 			return input.map(m -> {
 				if (!m.equals("fail")) {
 					return m.toUpperCase();
-				} else {
+				}
+				else {
 					throw new RuntimeException();
 				}
 			});
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -147,15 +156,18 @@ public class StreamListenerReactiveReturnWithFailureTests {
 	public static class ReactorTestReturnWithFailure4 {
 
 		@StreamListener
-		public @SendTo(Processor.OUTPUT)
-		Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
+		public @SendTo(Processor.OUTPUT) Flux<String> receive(
+				@Input(Processor.INPUT) Flux<String> input) {
 			return input.map(m -> {
 				if (!m.equals("fail")) {
 					return m.toUpperCase();
-				} else {
+				}
+				else {
 					throw new RuntimeException();
 				}
 			});
 		}
+
 	}
+
 }

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

@@ -57,25 +57,29 @@ public class StreamListenerReactiveReturnWithMessageTests {
 
 	@Parameterized.Parameters
 	public static Collection<?> InputConfigs() {
-		return Arrays.asList(ReactorTestReturnWithMessage1.class, ReactorTestReturnWithMessage2.class,
-				ReactorTestReturnWithMessage3.class, ReactorTestReturnWithMessage4.class);
+		return Arrays.asList(ReactorTestReturnWithMessage1.class,
+				ReactorTestReturnWithMessage2.class, ReactorTestReturnWithMessage3.class,
+				ReactorTestReturnWithMessage4.class);
 	}
 
 	@SuppressWarnings("unchecked")
-	private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
+	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());
+		processor.input().send(MessageBuilder.withPayload(sentPayload)
+				.setHeader("contentType", "text/plain").build());
 		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		Message<String> result = (Message<String>) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		Message<String> result = (Message<String>) messageCollector
+				.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
 		assertThat(result).isNotNull();
 		assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
 	}
 
 	@Test
 	public void testReturnWithMessage() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0",
-				"--spring.jmx.enabled=false",
+		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);
@@ -87,9 +91,11 @@ public class StreamListenerReactiveReturnWithMessageTests {
 	public static class ReactorTestReturnWithMessage1 {
 
 		@StreamListener
-		public @Output(Processor.OUTPUT) Flux<String> receive(@Input(Processor.INPUT) Flux<Message<String>> input) {
+		public @Output(Processor.OUTPUT) Flux<String> receive(
+				@Input(Processor.INPUT) Flux<Message<String>> input) {
 			return input.map(m -> m.getPayload().toUpperCase());
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -97,9 +103,11 @@ public class StreamListenerReactiveReturnWithMessageTests {
 	public static class ReactorTestReturnWithMessage2 {
 
 		@StreamListener(Processor.INPUT)
-		public @Output(Processor.OUTPUT) Flux<String> receive(Flux<Message<String>> input) {
+		public @Output(Processor.OUTPUT) Flux<String> receive(
+				Flux<Message<String>> input) {
 			return input.map(m -> m.getPayload().toUpperCase());
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -107,9 +115,11 @@ public class StreamListenerReactiveReturnWithMessageTests {
 	public static class ReactorTestReturnWithMessage3 {
 
 		@StreamListener(Processor.INPUT)
-		public @SendTo(Processor.OUTPUT) Flux<String> receive(Flux<Message<String>> input) {
+		public @SendTo(Processor.OUTPUT) Flux<String> receive(
+				Flux<Message<String>> input) {
 			return input.map(m -> m.getPayload().toUpperCase());
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -117,8 +127,11 @@ public class StreamListenerReactiveReturnWithMessageTests {
 	public static class ReactorTestReturnWithMessage4 {
 
 		@StreamListener
-		public @SendTo(Processor.OUTPUT) Flux<String> receive(@Input(Processor.INPUT) Flux<Message<String>> input) {
+		public @SendTo(Processor.OUTPUT) Flux<String> receive(
+				@Input(Processor.INPUT) Flux<Message<String>> input) {
 			return input.map(m -> m.getPayload().toUpperCase());
 		}
+
 	}
+
 }

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

@@ -61,22 +61,24 @@ public class StreamListenerReactiveReturnWithPojoTests {
 
 	@Parameterized.Parameters
 	public static Collection<?> InputConfigs() {
-		return Arrays.asList(ReactorTestReturnWithPojo1.class, ReactorTestReturnWithPojo2.class,
-				ReactorTestReturnWithPojo3.class, ReactorTestReturnWithPojo4.class);
+		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");
+		ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
+				"--server.port=0", "--spring.jmx.enabled=false");
 		Processor processor = context.getBean(Processor.class);
 		processor.input().send(MessageBuilder.withPayload("{\"message\":\"helloPojo\"}")
 				.setHeader("contentType", "application/json").build());
 		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		Message<String> result = (Message<String>) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		Message<String> result = (Message<String>) messageCollector
+				.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
 		assertThat(result).isNotNull();
-		BarPojo barPojo = mapper.readValue(result.getPayload(),BarPojo.class);
+		BarPojo barPojo = this.mapper.readValue(result.getPayload(), BarPojo.class);
 		assertThat(barPojo.getBarMessage()).isEqualTo("helloPojo");
 		context.close();
 	}
@@ -86,9 +88,11 @@ public class StreamListenerReactiveReturnWithPojoTests {
 	public static class ReactorTestReturnWithPojo1 {
 
 		@StreamListener
-		public @Output(Processor.OUTPUT) Flux<BarPojo> receive(@Input(Processor.INPUT) Flux<FooPojo> input) {
+		public @Output(Processor.OUTPUT) Flux<BarPojo> receive(
+				@Input(Processor.INPUT) Flux<FooPojo> input) {
 			return input.map(m -> new BarPojo(m.getMessage()));
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -99,6 +103,7 @@ public class StreamListenerReactiveReturnWithPojoTests {
 		public @Output(Processor.OUTPUT) Flux<BarPojo> receive(Flux<FooPojo> input) {
 			return input.map(m -> new BarPojo(m.getMessage()));
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -109,6 +114,7 @@ public class StreamListenerReactiveReturnWithPojoTests {
 		public @SendTo(Processor.OUTPUT) Flux<BarPojo> receive(Flux<FooPojo> input) {
 			return input.map(m -> new BarPojo(m.getMessage()));
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -116,9 +122,11 @@ public class StreamListenerReactiveReturnWithPojoTests {
 	public static class ReactorTestReturnWithPojo4 {
 
 		@StreamListener
-		public @SendTo(Processor.OUTPUT) Flux<BarPojo> receive(@Input(Processor.INPUT) Flux<FooPojo> input) {
+		public @SendTo(Processor.OUTPUT) Flux<BarPojo> receive(
+				@Input(Processor.INPUT) Flux<FooPojo> input) {
 			return input.map(m -> new BarPojo(m.getMessage()));
 		}
+
 	}
 
 	public static class FooPojo {
@@ -126,12 +134,13 @@ public class StreamListenerReactiveReturnWithPojoTests {
 		private String message;
 
 		public String getMessage() {
-			return message;
+			return this.message;
 		}
 
 		public void setMessage(String message) {
 			this.message = message;
 		}
+
 	}
 
 	public static class BarPojo {
@@ -144,11 +153,13 @@ public class StreamListenerReactiveReturnWithPojoTests {
 		}
 
 		public String getBarMessage() {
-			return barMessage;
+			return this.barMessage;
 		}
 
 		public void setBarMessage(String barMessage) {
 			this.barMessage = barMessage;
 		}
+
 	}
+
 }

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

@@ -47,20 +47,24 @@ import static org.springframework.cloud.stream.binding.StreamListenerErrorMessag
 public class StreamListenerWildCardFluxInputOutputArgsWithMessageTests {
 
 	@SuppressWarnings("unchecked")
-	private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
+	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());
+		processor.input().send(MessageBuilder.withPayload(sentPayload)
+				.setHeader("contentType", "text/plain").build());
 		MessageCollector messageCollector = context.getBean(MessageCollector.class);
-		Message<String> result = (Message<String>) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		Message<String> result = (Message<String>) messageCollector
+				.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
 		assertThat(result).isNotNull();
 		assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
 	}
 
 	@Test
 	public void testWildCardFluxInputOutputArgsWithMessage() throws Exception {
-		ConfigurableApplicationContext context = SpringApplication
-				.run(TestWildCardFluxInputOutputArgsWithMessage1.class, "--server.port=0","--spring.cloud.stream.bindings.output.contentType=text/plain");
+		ConfigurableApplicationContext context = SpringApplication.run(
+				TestWildCardFluxInputOutputArgsWithMessage1.class, "--server.port=0",
+				"--spring.cloud.stream.bindings.output.contentType=text/plain");
 		sendMessageAndValidate(context);
 		context.close();
 	}
@@ -68,18 +72,21 @@ public class StreamListenerWildCardFluxInputOutputArgsWithMessageTests {
 	@Test
 	public void testInputAsStreamListenerAndOutputAsParameterUsage() {
 		try {
-			SpringApplication.run(TestWildCardFluxInputOutputArgsWithMessage2.class, "--server.port=0");
+			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);
+			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");
+			SpringApplication.run(TestWildCardFluxInputOutputArgsWithMessage3.class,
+					"--server.port=0");
 			fail("Expected exception: " + INVALID_DECLARATIVE_METHOD_PARAMETERS);
 		}
 		catch (Exception e) {
@@ -94,8 +101,10 @@ public class StreamListenerWildCardFluxInputOutputArgsWithMessageTests {
 		@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()));
+			output.send(input.map(
+					m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -104,8 +113,10 @@ public class StreamListenerWildCardFluxInputOutputArgsWithMessageTests {
 
 		@StreamListener(Processor.INPUT)
 		public void receive(Flux<?> input, @Output(Processor.OUTPUT) FluxSender output) {
-			output.send(input.map(m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
+			output.send(input.map(
+					m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
 		}
+
 	}
 
 	@EnableBinding(Processor.class)
@@ -115,7 +126,10 @@ public class StreamListenerWildCardFluxInputOutputArgsWithMessageTests {
 		@StreamListener(Processor.INPUT)
 		@SendTo(Processor.OUTPUT)
 		public void receive(Flux<?> input, FluxSender output) {
-			output.send(input.map(m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
+			output.send(input.map(
+					m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
 		}
+
 	}
+
 }