Initial support for Reactive API

Fixes #520
Fixes #458 (for non-reactive binders)

Adds spring-cloud-stream-reactive module.
Introduces support for declarative @StreamListener and @Input and @Output annotated parameters.
Add StreamListenerArgumentAdapter and StreamListenerResultAdapter for wrapping bindable inputs
and outputs when passing arguments to declarative @StreamListener.
Adds support for using reactive types (Flux/Observable) with traditional binders.
Introduce FluxSender and ObservableSender for handling multiple streaming outputs per method.
Ensure that errors are caught and logged.

Fixing constructor assertions and Javadoc

Addressing PR comments

- rework @Input/@Output parameter validation
- renamed StreamListenerArgumentAdapter to StreamListenerParameterAdapter
- ensure that parameter direction is accounted for in the current adapters

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.reactive;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * Used for {@link org.springframework.cloud.stream.annotation.StreamListener} arguments annotated with {@link
+ * org.springframework.cloud.stream.annotation.Output}.
+ * @author Marius Bogoevici
+ */
+public interface FluxSender {
+
+	/**
+	 * Streams the {@link reactor.core.publisher.Flux} through the bound
+	 * element corresponding to the {@link org.springframework.cloud.stream.annotation.Output} annotation of the
+	 * argument.
+	 * @param flux a {@link Flux} that will be streamed through the bound element
+	 * @return a {@link Mono} representing the result of sending the flux (completion or error)
+	 */
+	Mono<Void> send(Flux<?> flux);
+}

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

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.reactive;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Flux;
+
+import org.springframework.cloud.stream.binding.StreamListenerResultAdapter;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * A {@link org.springframework.cloud.stream.binding.StreamListenerResultAdapter} from a {@link Flux}
+ * return type to a bound {@link MessageChannel}.
+ * @author Marius Bogoevici
+ */
+public class FluxToMessageChannelResultAdapter
+		implements StreamListenerResultAdapter<Flux<?>, MessageChannel> {
+
+	private Log log = LogFactory.getLog(FluxToMessageChannelResultAdapter.class);
+
+	@Override
+	public boolean supports(Class<?> resultType, Class<?> boundType) {
+		return Flux.class.isAssignableFrom(resultType) && MessageChannel.class.isAssignableFrom(boundType);
+	}
+
+	public void adapt(Flux<?> streamListenerResult, MessageChannel boundElement) {
+		streamListenerResult
+				.doOnError(e -> this.log.error("Error while processing result", e))
+				.retry()
+				.subscribe(
+						result -> boundElement.send(result instanceof Message<?> ? (Message<?>) result
+								: MessageBuilder.withPayload(result).build()));
+	}
+}

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

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.reactive;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.MonoProcessor;
+
+import org.springframework.cloud.stream.annotation.Output;
+import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
+import org.springframework.core.MethodParameter;
+import org.springframework.core.ResolvableType;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * Adapts an {@link org.springframework.cloud.stream.annotation.Output} annotated
+ * {@link FluxSender} to an outbound {@link MessageChannel}.
+ * @author Marius Bogoevici
+ */
+public class MessageChannelToFluxSenderParameterAdapter
+		implements StreamListenerParameterAdapter<FluxSender, MessageChannel> {
+
+	private Log log = LogFactory.getLog(MessageChannelToFluxSenderParameterAdapter.class);
+
+	@Override
+	public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
+		ResolvableType type = ResolvableType.forMethodParameter(methodParameter);
+		return MessageChannel.class.isAssignableFrom(boundElementType)
+				&& methodParameter.getParameterAnnotation(Output.class) != null
+				&& FluxSender.class.isAssignableFrom(type.getRawClass());
+	}
+
+	@Override
+	public FluxSender adapt(MessageChannel boundElement, MethodParameter parameter) {
+		return resultPublisher -> {
+			MonoProcessor<Void> sendResult = MonoProcessor.create();
+			// add error handling and reconnect in the event of an error
+			resultPublisher
+					.doOnError(e -> this.log.error("Error during processing: ", e))
+					.retry()
+					.subscribe(
+							result -> boundElement.send(result instanceof Message<?> ? (Message<?>) result :
+									MessageBuilder.withPayload(result).build()), e -> sendResult.onError(e),
+							() -> sendResult.onComplete());
+			return sendResult;
+		};
+	}
+}

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

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.reactive;
+
+import reactor.core.publisher.Flux;
+
+import org.springframework.cloud.stream.annotation.Input;
+import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
+import org.springframework.core.MethodParameter;
+import org.springframework.core.ResolvableType;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.converter.CompositeMessageConverter;
+import org.springframework.util.Assert;
+
+/**
+ * Adapts an {@link org.springframework.cloud.stream.annotation.Input} annotated
+ * {@link MessageChannel} to a {@link Flux}.
+ * @author Marius Bogoevici
+ */
+public class MessageChannelToInputFluxParameterAdapter
+		implements StreamListenerParameterAdapter<Flux<?>, SubscribableChannel> {
+
+	private final CompositeMessageConverter messageConverter;
+
+	public MessageChannelToInputFluxParameterAdapter(CompositeMessageConverter messageConverter) {
+		Assert.notNull(messageConverter, "cannot not be null");
+		this.messageConverter = messageConverter;
+	}
+
+	@Override
+	public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
+		return SubscribableChannel.class.isAssignableFrom(boundElementType)
+				&& methodParameter.getParameterAnnotation(Input.class) != null
+				&& Flux.class.isAssignableFrom(methodParameter.getParameterType());
+	}
+
+	@Override
+	public Flux<?> adapt(final SubscribableChannel boundElement, MethodParameter parameter) {
+		ResolvableType resolvableType = ResolvableType.forMethodParameter(parameter);
+		Class<?> argumentClass = resolvableType.getGeneric(0).getRawClass();
+		final Object monitor = new Object();
+		if (Message.class.isAssignableFrom(argumentClass)) {
+			return Flux.create(emitter -> {
+				MessageHandler messageHandler = message -> {
+					synchronized (monitor) {
+						emitter.next(message);
+					}
+				};
+				boundElement.subscribe(messageHandler);
+				emitter.setCancellation(() -> boundElement.unsubscribe(messageHandler));
+			});
+		}
+		else {
+			return Flux.create(emitter -> {
+				MessageHandler messageHandler = message -> {
+					synchronized (monitor) {
+						if (argumentClass.isAssignableFrom(message.getPayload().getClass())) {
+							emitter.next(message.getPayload());
+						}
+						else {
+							emitter.next(this.messageConverter.fromMessage(message, argumentClass));
+						}
+					}
+				};
+				boundElement.subscribe(messageHandler);
+				emitter.setCancellation(() -> boundElement.unsubscribe(messageHandler));
+			});
+		}
+	}
+}

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

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.reactive;
+
+import reactor.adapter.RxJava1Adapter;
+import rx.Observable;
+
+import org.springframework.cloud.stream.annotation.Input;
+import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
+import org.springframework.core.MethodParameter;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.util.Assert;
+
+/**
+ * Adapts an {@link org.springframework.cloud.stream.annotation.Input} annotated
+ * {@link MessageChannel} to an {@link Observable}.
+ * @author Marius Bogoevici
+ */
+public class MessageChannelToInputObservableParameterAdapter
+		implements StreamListenerParameterAdapter<Observable<?>, SubscribableChannel> {
+
+	private final MessageChannelToInputFluxParameterAdapter messageChannelToInputFluxArgumentAdapter;
+
+	public MessageChannelToInputObservableParameterAdapter(
+			MessageChannelToInputFluxParameterAdapter messageChannelToInputFluxArgumentAdapter) {
+		Assert.notNull(messageChannelToInputFluxArgumentAdapter, "cannot be null");
+		this.messageChannelToInputFluxArgumentAdapter = messageChannelToInputFluxArgumentAdapter;
+	}
+
+	public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
+		return SubscribableChannel.class.isAssignableFrom(boundElementType)
+				&& methodParameter.getParameterAnnotation(Input.class) != null
+				&& Observable.class.isAssignableFrom(methodParameter.getParameterType());
+	}
+
+	@Override
+	public Observable<?> adapt(final SubscribableChannel boundElement, MethodParameter parameter) {
+		return RxJava1Adapter.publisherToObservable(
+				this.messageChannelToInputFluxArgumentAdapter.adapt(boundElement, parameter));
+	}
+}

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

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.reactive;
+
+import reactor.adapter.RxJava1Adapter;
+import rx.Observable;
+import rx.Single;
+
+import org.springframework.cloud.stream.annotation.Output;
+import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
+import org.springframework.core.MethodParameter;
+import org.springframework.core.ResolvableType;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
+
+/**
+ * Adapts an {@link org.springframework.cloud.stream.annotation.Output} annotated
+ * {@link ObservableSender} to an outbound {@link MessageChannel}.
+ * @author Marius Bogoevici
+ */
+public class MessageChannelToObservableSenderParameterAdapter implements
+		StreamListenerParameterAdapter<ObservableSender, MessageChannel> {
+
+	private final MessageChannelToFluxSenderParameterAdapter messageChannelToFluxSenderArgumentAdapter;
+
+	public MessageChannelToObservableSenderParameterAdapter(
+			MessageChannelToFluxSenderParameterAdapter messageChannelToFluxSenderArgumentAdapter) {
+		Assert.notNull(messageChannelToFluxSenderArgumentAdapter, "cannot be null");
+		this.messageChannelToFluxSenderArgumentAdapter = messageChannelToFluxSenderArgumentAdapter;
+	}
+
+	@Override
+	public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
+		ResolvableType type = ResolvableType.forMethodParameter(methodParameter);
+		return MessageChannel.class.isAssignableFrom(boundElementType)
+				&& methodParameter.getParameterAnnotation(Output.class) != null
+				&& ObservableSender.class.isAssignableFrom(type.getRawClass());
+	}
+
+	@Override
+	public ObservableSender adapt(MessageChannel boundElement, MethodParameter parameter) {
+		return new ObservableSender() {
+
+			private FluxSender fluxSender = MessageChannelToObservableSenderParameterAdapter.this
+					.messageChannelToFluxSenderArgumentAdapter
+					.adapt(boundElement, parameter);
+
+			@Override
+			public Single<Void> send(Observable<?> observable) {
+				return RxJava1Adapter.publisherToSingle(
+						this.fluxSender.send(RxJava1Adapter.observableToFlux(observable)));
+			}
+		};
+	}
+}

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.reactive;
+
+import rx.Observable;
+import rx.Single;
+
+/**
+ * Used for {@link org.springframework.cloud.stream.annotation.StreamListener} arguments annotated with {@link
+ * org.springframework.cloud.stream.annotation.Output}.
+ * @author Marius Bogoevici
+ */
+public interface ObservableSender {
+
+	/**
+	 * Streams the {@link Observable} through the bound
+	 * element corresponding to the {@link org.springframework.cloud.stream.annotation.Output} annotation of the
+	 * argument.
+	 * @param observable an {@link Observable} that will be streamed through the bound element
+	 * @return a {@link Single} representing the result of an operation
+	 */
+	Single<Void> send(Observable<?> observable);
+}

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

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.reactive;
+
+import reactor.adapter.RxJava1Adapter;
+import rx.Observable;
+
+import org.springframework.cloud.stream.binding.StreamListenerResultAdapter;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link StreamListenerResultAdapter} from an {@link Observable}
+ * return type to a bound {@link MessageChannel}.
+ * @author Marius Bogoevici
+ */
+public class ObservableToMessageChannelResultAdapter
+		implements StreamListenerResultAdapter<Observable<?>, MessageChannel> {
+
+	private FluxToMessageChannelResultAdapter fluxToMessageChannelResultAdapter;
+
+	public ObservableToMessageChannelResultAdapter(
+			FluxToMessageChannelResultAdapter fluxToMessageChannelResultAdapter) {
+		Assert.notNull(fluxToMessageChannelResultAdapter, "cannot be null");
+		this.fluxToMessageChannelResultAdapter = fluxToMessageChannelResultAdapter;
+	}
+
+	@Override
+	public boolean supports(Class<?> resultType, Class<?> boundType) {
+		return Observable.class.isAssignableFrom(resultType)
+				&& MessageChannel.class.isAssignableFrom(boundType);
+	}
+
+	public void adapt(Observable<?> streamListenerResult, MessageChannel boundElement) {
+		this.fluxToMessageChannelResultAdapter.adapt(RxJava1Adapter.observableToFlux(streamListenerResult),
+				boundElement);
+	}
+}

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

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * 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.ConditionalOnClass;
+import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author Marius Bogoevici
+ */
+@Configuration
+public class ReactiveSupportAutoConfiguration {
+
+	@Bean
+	public MessageChannelToInputFluxParameterAdapter messageChannelToInputFluxArgumentAdapter(
+			CompositeMessageConverterFactory compositeMessageConverterFactory) {
+		return new MessageChannelToInputFluxParameterAdapter(
+				compositeMessageConverterFactory.getMessageConverterForAllRegistered());
+	}
+
+	@Bean
+	public MessageChannelToFluxSenderParameterAdapter messageChannelToFluxSenderArgumentAdapter() {
+		return new MessageChannelToFluxSenderParameterAdapter();
+	}
+
+	@Bean
+	public FluxToMessageChannelResultAdapter fluxToMessageChannelResultAdapter() {
+		return new FluxToMessageChannelResultAdapter();
+	}
+
+	@Configuration
+	@ConditionalOnClass(name = "rx.Observable")
+	public static class RxJava1SupportConfiguration {
+
+		@Bean
+		public MessageChannelToInputObservableParameterAdapter messageChannelToInputObservableArgumentAdapter(
+				MessageChannelToInputFluxParameterAdapter messageChannelToFluxArgumentAdapter) {
+			return new MessageChannelToInputObservableParameterAdapter(messageChannelToFluxArgumentAdapter);
+		}
+
+		@Bean
+		public MessageChannelToObservableSenderParameterAdapter messageChannelToObservableSenderArgumentAdapter(
+				MessageChannelToFluxSenderParameterAdapter messageChannelToFluxSenderArgumentAdapter) {
+			return new MessageChannelToObservableSenderParameterAdapter(messageChannelToFluxSenderArgumentAdapter);
+		}
+
+		@Bean
+		public ObservableToMessageChannelResultAdapter
+		observableToMessageChannelResultAdapter(
+				FluxToMessageChannelResultAdapter fluxToMessageChannelResultAdapter) {
+			return new ObservableToMessageChannelResultAdapter(fluxToMessageChannelResultAdapter);
+		}
+	}
+}

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

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

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

@@ -0,0 +1,274 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * 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.support.MessageBuilder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class StreamListenerReactorTests {
+
+	@Test
+	public void testInputOutputArgs() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgs.class, "--server.port=0");
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	private void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
+		@SuppressWarnings("unchecked")
+		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 = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		assertThat(result).isNotNull();
+		assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
+	}
+
+	private void sendFailingMessage(ConfigurableApplicationContext context) throws InterruptedException {
+		@SuppressWarnings("unchecked")
+		Processor processor = context.getBean(Processor.class);
+		processor.input().send(MessageBuilder.withPayload("fail").setHeader("contentType", "text/plain").build());
+	}
+
+	@Test
+	public void testInputOutputArgsWithMessage() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithMessage.class,
+				"--server.port=0");
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	@Test
+	public void testInputOutputArgsWithFluxSender() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithFluxSender.class,
+				"--server.port=0");
+		// send multiple message
+		sendMessageAndValidate(context);
+		sendMessageAndValidate(context);
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	@Test
+	public void testInputOutputArgsWithFluxSenderAndFailure() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithFluxSenderAndFailure.class, "--server.port=0");
+		sendMessageAndValidate(context);
+		sendFailingMessage(context);
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	@Test
+	public void testReturn() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestReturn.class, "--server.port=0");
+		sendMessageAndValidate(context);
+		sendMessageAndValidate(context);
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	@Test
+	public void testReturnWithFailure() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithFailure.class, "--server.port=0");
+		sendMessageAndValidate(context);
+		sendFailingMessage(context);
+		sendMessageAndValidate(context);
+		sendFailingMessage(context);
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	@Test
+	public void testReturnWithMessage() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithMessage.class, "--server.port=0");
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	@Test
+	public void testReturnWithPojo() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithPojo.class, "--server.port=0");
+		@SuppressWarnings("unchecked")
+		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 = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		assertThat(result).isNotNull();
+		assertThat(result.getPayload()).isInstanceOf(BarPojo.class);
+		assertThat(((BarPojo) result.getPayload()).getBarMessage()).isEqualTo("helloPojo");
+		context.close();
+	}
+
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestInputOutputArgs {
+
+		@StreamListener
+		public void receive(@Input(Processor.INPUT) Flux<String> input, @Output(Processor.OUTPUT) FluxSender output) {
+			output.send(input.map(m -> m.toUpperCase()));
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestInputOutputArgsWithMessage {
+
+		@StreamListener
+		public void receive(@Input(Processor.INPUT) Flux<Message<String>> input,
+				@Output(Processor.OUTPUT) FluxSender output) {
+			output.send(input.map(m -> MessageBuilder.withPayload(m.getPayload().toUpperCase()).build()));
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestInputOutputArgsWithFluxSender {
+		@StreamListener
+		public void receive(@Input(Processor.INPUT) Flux<Message<?>> input, @Output(Processor.OUTPUT) FluxSender output) {
+			output.send(input
+					.map(m -> m.getPayload().toString().toUpperCase())
+					.map(o -> MessageBuilder.withPayload(o).build()));
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestInputOutputArgsWithFluxSenderAndFailure {
+		@StreamListener
+		public void receive(@Input(Processor.INPUT) Flux<Message<?>> 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()));
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestReturn {
+
+		@StreamListener
+		public
+		@Output(Processor.OUTPUT)
+		Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
+			return input.map(m -> m.toUpperCase());
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestReturnWithFailure {
+
+		@StreamListener
+		public
+		@Output(Processor.OUTPUT)
+		Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
+			return input.map(m -> {
+				if (!m.equals("fail")) {
+					return m.toUpperCase();
+				}
+				else {
+					throw new RuntimeException();
+				}
+			});
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestReturnWithMessage {
+
+		@StreamListener
+		public
+		@Output(Processor.OUTPUT)
+		Flux<String> receive(@Input(Processor.INPUT) Flux<Message<String>> input) {
+			return input.map(m -> m.getPayload().toUpperCase());
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestReturnWithPojo {
+
+		@StreamListener
+		public
+		@Output(Processor.OUTPUT)
+		Flux<BarPojo> receive(@Input(Processor.INPUT) Flux<FooPojo> input) {
+			return input.map(m -> new BarPojo(m.getMessage()));
+		}
+	}
+
+	public static class FooPojo {
+
+		private String message;
+
+		public String getMessage() {
+			return message;
+		}
+
+		public void setMessage(String message) {
+			this.message = message;
+		}
+	}
+
+	public static class BarPojo {
+
+		private String barMessage;
+
+		public BarPojo(String barMessage) {
+			this.barMessage = barMessage;
+		}
+
+		public String getBarMessage() {
+			return barMessage;
+		}
+
+		public void setBarMessage(String barMessage) {
+			this.barMessage = barMessage;
+		}
+	}
+}

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

@@ -0,0 +1,245 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * 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 rx.Observable;
+
+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
+ */
+public class StreamListenerRxJava1Tests {
+
+	@Test
+	public void testInputOutputArgs() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgs.class, "--server.port=0");
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	private void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
+		@SuppressWarnings("unchecked")
+		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 = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		assertThat(result).isNotNull();
+		assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
+	}
+
+	private void sendFailingMessage(ConfigurableApplicationContext context) throws InterruptedException {
+		@SuppressWarnings("unchecked")
+		Processor processor = context.getBean(Processor.class);
+		processor.input().send(MessageBuilder.withPayload("fail").setHeader("contentType", "text/plain").build());
+	}
+
+	@Test
+	public void testInputOutputArgsWithMessage() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithMessage.class,
+				"--server.port=0");
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	@Test
+	public void testInputOutputArgsWithObservableSender() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithObservableSender.class,
+				"--server.port=0");
+		// send multiple message
+		sendMessageAndValidate(context);
+		sendMessageAndValidate(context);
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	@Test
+	public void testReturn() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestReturn.class, "--server.port=0");
+		sendMessageAndValidate(context);
+		sendMessageAndValidate(context);
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	@Test
+	public void testReturnWithFailure() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithFailure.class, "--server.port=0");
+		sendMessageAndValidate(context);
+		sendFailingMessage(context);
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	@Test
+	public void testReturnWithMessage() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithMessage.class, "--server.port=0");
+		sendMessageAndValidate(context);
+		context.close();
+	}
+
+	@Test
+	public void testReturnWithPojo() throws Exception {
+		ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithPojo.class, "--server.port=0");
+		@SuppressWarnings("unchecked")
+		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 = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
+		assertThat(result).isNotNull();
+		assertThat(result.getPayload()).isInstanceOf(BarPojo.class);
+		assertThat(((BarPojo) result.getPayload()).getBarMessage()).isEqualTo("helloPojo");
+		context.close();
+	}
+
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestInputOutputArgs {
+
+		@StreamListener
+		public void receive(@Input(Processor.INPUT) Observable<String> input, @Output(Processor.OUTPUT) ObservableSender output) {
+			output.send(input.map(m -> m.toUpperCase()));
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestInputOutputArgsWithMessage {
+
+		@StreamListener
+		public void receive(@Input(Processor.INPUT) Observable<Message<String>> input,
+				@Output(Processor.OUTPUT) ObservableSender output) {
+			output.send(input.map(m -> MessageBuilder.withPayload(m.getPayload().toUpperCase()).build()));
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestInputOutputArgsWithObservableSender {
+		@StreamListener
+		public void receive(@Input(Processor.INPUT) Observable<Message<?>> input, @Output(Processor.OUTPUT)
+				ObservableSender output) {
+			output.send(input
+					.map(m -> m.getPayload().toString().toUpperCase())
+					.map(o -> MessageBuilder.withPayload(o).build()));
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestReturn {
+
+		@StreamListener
+		public
+		@Output(Processor.OUTPUT)
+		Observable<String> receive(@Input(Processor.INPUT) Observable<String> input) {
+			return input.map(m -> m.toUpperCase());
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestReturnWithFailure {
+
+		@StreamListener
+		public
+		@Output(Processor.OUTPUT)
+		Observable<String> receive(@Input(Processor.INPUT) Observable<String> input) {
+			return input.map(m -> {
+				if (!m.equals("fail")) {
+					return m.toUpperCase();
+				}
+				else {
+					throw new RuntimeException();
+				}
+			});
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestReturnWithMessage {
+
+		@StreamListener
+		public
+		@Output(Processor.OUTPUT)
+		Observable<String> receive(@Input(Processor.INPUT) Observable<Message<String>> input) {
+			return input.map(m -> m.getPayload().toUpperCase());
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestReturnWithPojo {
+
+		@StreamListener
+		public
+		@Output(Processor.OUTPUT)
+		Observable<BarPojo> receive(@Input(Processor.INPUT) Observable<FooPojo> input) {
+			return input.map(m -> new BarPojo(m.getMessage()));
+		}
+	}
+
+	public static class FooPojo {
+
+		private String message;
+
+		public String getMessage() {
+			return message;
+		}
+
+		public void setMessage(String message) {
+			this.message = message;
+		}
+	}
+
+	public static class BarPojo {
+
+		private String barMessage;
+
+		public BarPojo(String barMessage) {
+			this.barMessage = barMessage;
+		}
+
+		public String getBarMessage() {
+			return barMessage;
+		}
+
+		public void setBarMessage(String barMessage) {
+			this.barMessage = barMessage;
+		}
+	}
+}