DSL-30: Add Reactive Streams support

build.gradle

@@ -57,13 +57,16 @@ ext {
 	jmsApiVersion = '1.1-rev-1'
 	jrubyVersion = '1.7.19'
 	jythonVersion = '2.5.3'
-	kafkaVersion = '0.8.1.1'
+	kafkaVersion = '0.8.2.1'
 	mailVersion = '1.5.2'
+	reactiveStreamsVersion = '1.0.0'
+	reactorVersion = '2.0.4.RELEASE'
 	scalaVersion = '2.10'
-	slf4jVersion = '1.7.11'
-	springIntegrationVersion = '4.1.3.RELEASE'
-	springIntegrationKafkaVersion = '1.1.1.RELEASE'
-	springBootVersion = '1.2.3.RELEASE'
+	slf4jVersion = '1.7.12'
+	springIntegrationVersion = '4.2.0.M2'
+	springIntegrationKafkaVersion = '1.2.0.RELEASE'
+	springBootVersion = '1.3.0.M1'
+	testNgVersion = '6.8.21'
 
 	linkHomepage = 'https://github.com/spring-projects/spring-integration-java-dsl'
 	linkCi = 'https://build.spring.io/browse/INTEXT-SIJD'
@@ -82,7 +85,7 @@ configurations {
 }
 
 dependencies {
-	compile "org.springframework.integration:spring-integration-core:$springIntegrationVersion"
+	compile("org.springframework.integration:spring-integration-core:$springIntegrationVersion")
 
 	['spring-integration-amqp'
 	 , 'spring-integration-event'
@@ -117,10 +120,13 @@ dependencies {
 		exclude group: 'com.yammer.metrics', module: 'metrics-annotation'
 		exclude group: 'org.apache.velocity', module: 'velocity'
 		exclude group: 'log4j', module: 'log4j'
+		exclude group: 'org.slf4j', module: 'slf4j-log4j12'
 	}
 	compile("javax.jms:jms-api:$jmsApiVersion", provided)
 	compile("javax.mail:javax.mail-api:$mailVersion", provided)
 
+	compile("org.reactivestreams:reactive-streams:$reactiveStreamsVersion", optional)
+
 	testCompile "org.springframework.integration:spring-integration-test:$springIntegrationVersion"
 	testCompile("org.springframework.boot:spring-boot-starter:$springBootVersion") {
 		exclude module: 'spring-boot-starter-logging'
@@ -129,9 +135,15 @@ dependencies {
 	testCompile "de.flapdoodle.embed:de.flapdoodle.embed.mongo:$embedMongoVersion"
 	testCompile "org.apache.ftpserver:ftpserver-core:$ftpServerVersion"
 	testCompile "org.apache.sshd:sshd-core:$apacheSshdVersion"
-	testCompile ("org.apache.kafka:kafka_$scalaVersion:$kafkaVersion:test") {
+	testCompile("org.apache.kafka:kafka_$scalaVersion:$kafkaVersion:test") {
 		exclude group: 'log4j', module: 'log4j'
+		exclude group: 'org.slf4j', module: 'slf4j-log4j12'
+	}
+	testCompile("org.reactivestreams:reactive-streams-tck:$reactiveStreamsVersion") {
+		exclude group: 'org.reactivestreams', module: 'reactive-streams-examples'
 	}
+	testCompile "io.projectreactor:reactor-stream:$reactorVersion"
+	testCompile "org.testng:testng:$testNgVersion"
 
 	testRuntime "org.slf4j:slf4j-log4j12:$slf4jVersion"
 	testRuntime "org.apache.activemq:activemq-broker:$activeMqVersion"

src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java

@@ -21,6 +21,9 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+import org.reactivestreams.Publisher;
 
 import org.springframework.aop.framework.Advised;
 import org.springframework.aop.support.AopUtils;
@@ -33,6 +36,7 @@
 import org.springframework.integration.channel.ChannelInterceptorAware;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.FixedSubscriberChannel;
+import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.channel.interceptor.WireTap;
 import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
 import org.springframework.integration.core.GenericSelector;
@@ -83,6 +87,7 @@
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.PollableChannel;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
@@ -2442,6 +2447,37 @@ public B gateway(IntegrationFlow flow, Consumer<GatewayEndpointSpec> endpointCon
 		return gateway(requestChannel, endpointConfigurer);
 	}
 
+	/**
+	 * Represent an Integration Flow as a Reactive Streams {@link Publisher} bean.
+	 * @param <T> the {@code payload} type
+	 * @return the Reactive Streams {@link Publisher}
+	 */
+	public <T> Publisher<Message<T>> toReactivePublisher() {
+		return toReactivePublisher(Executors.newSingleThreadExecutor());
+	}
+
+	/**
+	 * Represent an Integration Flow as a Reactive Streams {@link Publisher} bean.
+	 * @param executor the managed {@link Executor} to be used for the background task to
+	 *                 poll messages from the {@link PollableChannel}.
+	 *                 Defaults to {@link Executors#newSingleThreadExecutor()}.
+	 * @param <T> the {@code payload} type
+	 * @return the Reactive Streams {@link Publisher}
+	 */
+	@SuppressWarnings("unchecked")
+	public <T> Publisher<Message<T>> toReactivePublisher(Executor executor) {
+		Assert.notNull(executor);
+		MessageChannel channelForPublisher = this.currentMessageChannel;
+		if (channelForPublisher == null) {
+			PublishSubscribeChannel publishSubscribeChannel = new PublishSubscribeChannel();
+			publishSubscribeChannel.setMinSubscribers(1);
+			channelForPublisher = publishSubscribeChannel;
+			channel(channelForPublisher);
+		}
+		get();
+		return new PublisherIntegrationFlow<T>(this.integrationComponents, channelForPublisher, executor);
+	}
+
 	private <S extends ConsumerEndpointSpec<S, ?>> B register(S endpointSpec, Consumer<S> endpointConfigurer) {
 		if (endpointConfigurer != null) {
 			endpointConfigurer.accept(endpointSpec);
@@ -2544,27 +2580,6 @@ protected final B _this() {
 		return (B) this;
 	}
 
-	private static boolean isLambda(Object o) {
-		Class<?> aClass = o.getClass();
-		return aClass.isSynthetic() && !aClass.isAnonymousClass() && !aClass.isLocalClass();
-	}
-
-	private static Object extractProxyTarget(Object target) {
-		if (!(target instanceof Advised)) {
-			return target;
-		}
-		Advised advised = (Advised) target;
-		if (advised.getTargetSource() == null) {
-			return null;
-		}
-		try {
-			return extractProxyTarget(advised.getTargetSource().getTarget());
-		}
-		catch (Exception e) {
-			throw new BeanCreationException("Could not extract target", e);
-		}
-	}
-
 	protected StandardIntegrationFlow get() {
 		if (this.currentMessageChannel instanceof FixedSubscriberChannelPrototype) {
 			throw new BeanCreationException("The 'currentMessageChannel' (" + this.currentMessageChannel +
@@ -2588,4 +2603,25 @@ else if (this.currentMessageChannel != null) {
 		return new StandardIntegrationFlow(this.integrationComponents);
 	}
 
+	private static boolean isLambda(Object o) {
+		Class<?> aClass = o.getClass();
+		return aClass.isSynthetic() && !aClass.isAnonymousClass() && !aClass.isLocalClass();
+	}
+
+	private static Object extractProxyTarget(Object target) {
+		if (!(target instanceof Advised)) {
+			return target;
+		}
+		Advised advised = (Advised) target;
+		if (advised.getTargetSource() == null) {
+			return null;
+		}
+		try {
+			return extractProxyTarget(advised.getTargetSource().getTarget());
+		}
+		catch (Exception e) {
+			throw new BeanCreationException("Could not extract target", e);
+		}
+	}
+
 }