Fixes according the latest Reactor

spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java

@@ -25,11 +25,10 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
-import reactor.core.flow.Cancellation;
 import reactor.core.publisher.EmitterProcessor;
-import reactor.core.scheduler.Scheduler;
 import reactor.core.subscriber.BaseSubscriber;
 import reactor.core.subscriber.SignalEmitter;
+import reactor.core.util.PlatformDependent;
 
 /**
  * @author Artem Bilan
@@ -42,7 +41,7 @@ public class ReactiveChannel implements MessageChannel, Publisher<Message<?>> {
 	private final SignalEmitter<Message<?>> emitter;
 
 	public ReactiveChannel() {
-		this(EmitterProcessor.async(SyncScheduler.INSTANCE));
+		this(EmitterProcessor.create(PlatformDependent.SMALL_BUFFER_SIZE, Integer.MAX_VALUE, false));
 	}
 
 	public ReactiveChannel(Processor<Message<?>, Message<?>> processor) {
@@ -83,37 +82,4 @@ public class ReactiveChannel implements MessageChannel, Publisher<Message<?>> {
 		this.processor.subscribe(subscriber);
 	}
 
-
-	private static final class SyncScheduler implements Scheduler {
-
-		private final static Scheduler INSTANCE = new SyncScheduler();
-
-		private final Worker worker = new Worker() {
-
-			@Override
-			public Cancellation schedule(Runnable task) {
-				task.run();
-				return () -> {
-				};
-			}
-
-			@Override
-			public void shutdown() {
-
-			}
-
-		};
-
-		@Override
-		public Cancellation schedule(Runnable task) {
-			return this.worker.schedule(task);
-		}
-
-		@Override
-		public Worker createWorker() {
-			return this.worker;
-		}
-
-	}
-
 }

spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * 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.
@@ -35,8 +35,6 @@ import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
-import reactor.core.publisher.Processors;
-
 /**
  * @author Artem Bilan
  * @since 5.0
@@ -71,7 +69,7 @@ public class ReactiveChannelTests {
 
 		@Bean
 		public MessageChannel reactiveChannel() {
-			return new ReactiveChannel(Processors.queue());
+			return new ReactiveChannel();
 		}
 
 		@ServiceActivator(inputChannel = "reactiveChannel")

spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java

@@ -127,7 +127,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.test.context.support.AnnotationConfigContextLoader;
 import org.springframework.util.MultiValueMap;
 
-import reactor.rx.Stream;
+import reactor.core.publisher.Flux;
 
 /**
  * @author Artem Bilan
@@ -617,11 +617,11 @@ public class EnableIntegrationTests {
 		final AtomicReference<List<Integer>> ref = new AtomicReference<List<Integer>>();
 		final CountDownLatch consumeLatch = new CountDownLatch(1);
 
-		Stream.just("1", "2", "3", "4", "5")
+		Flux.just("1", "2", "3", "4", "5")
 				.map(Integer::parseInt)
 				.flatMap(this.testGateway::multiply)
 				.toList()
-				.doOnSuccess(integers -> {
+				.subscribe(integers -> {
 					ref.set(integers);
 					consumeLatch.countDown();
 				});
@@ -1349,7 +1349,7 @@ public class EnableIntegrationTests {
 	@Target({ ElementType.TYPE, ElementType.ANNOTATION_TYPE })
 	@Retention(RetentionPolicy.RUNTIME)
 	@MessagingGateway(defaultRequestChannel = "gatewayChannel",
-			defaultRequestTimeout="${default.request.timeout:12300}", defaultReplyTimeout="#{13400}",
+			defaultRequestTimeout = "${default.request.timeout:12300}", defaultReplyTimeout = "#{13400}",
 			defaultHeaders = @GatewayHeader(name = "foo", value = "FOO"))
 	public @interface TestMessagingGateway {
 

spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests.java

@@ -60,7 +60,8 @@ import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.test.context.support.GenericXmlContextLoader;
 
-import reactor.rx.Stream;
+import reactor.core.publisher.Flux;
+
 
 /**
  * @author Artem Bilan
@@ -145,7 +146,7 @@ public class RoutingSlipTests {
 
 	public static class TestRoutingSlipRoutePojo {
 
-		final String[] channels = {"channel2", "channel3"};
+		final String[] channels = { "channel2", "channel3" };
 
 		private int i = 0;
 
@@ -189,13 +190,15 @@ public class RoutingSlipTests {
 		public RoutingSlipRouteStrategy routeStrategy() {
 			return (requestMessage, reply) -> requestMessage.getPayload() instanceof String
 					? new FixedSubscriberChannel(m ->
-					Stream.just((String) m.getPayload())
+					Flux.just((String) m.getPayload())
 							.map(String::toUpperCase)
-							.consume(v -> messagingTemplate().convertAndSend(resultsChannel(), v)))
+							.subscribe(v -> messagingTemplate()
+									.convertAndSend(resultsChannel(), v)))
 					: new FixedSubscriberChannel(m ->
-					Stream.just((Integer) m.getPayload())
+					Flux.just((Integer) m.getPayload())
 							.map(v -> v * 2)
-							.consume(v -> messagingTemplate().convertAndSend(resultsChannel(), v)));
+							.subscribe(v -> messagingTemplate()
+									.convertAndSend(resultsChannel(), v)));
 		}
 
 		@Bean