Rename ReactiveChannel

- as discussed last week

TODO:
- should we take the channel outside of the `AbstractMessageChannel` hierarchy?
  - avoid blocking interceptors
  - we would lose channel metrics though
- rename `ReactiveConsumer` ?

Polishing some missed renaming

Author: garyrussell

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

@@ -32,34 +32,34 @@
 import reactor.core.publisher.FluxSink;
 
 /**
+ * The {@link AbstractMessageChannel} implementation for the
+ * Reactive Streams {@link Publisher} based on the Project Reactor {@link FluxProcessor}.
+ *
  * @author Artem Bilan
  * @author Gary Russell
  *
  * @since 5.0
  */
-public class ReactiveChannel extends AbstractMessageChannel
-		implements Publisher<Message<?>>, ReactiveSubscribableChannel {
+public class FluxMessageChannel extends AbstractMessageChannel
+		implements Publisher<Message<?>>, FluxSubscribableChannel {
 
 	private final List<Subscriber<? super Message<?>>> subscribers = new ArrayList<>();
 
 	private final List<Publisher<Message<?>>> publishers = new CopyOnWriteArrayList<>();
 
 	private final FluxProcessor<Message<?>, Message<?>> processor;
 
-	private final Flux<Message<?>> flux;
-
 	private final FluxSink<Message<?>> sink;
 
 	private volatile boolean upstreamSubscribed;
 
-	public ReactiveChannel() {
+	public FluxMessageChannel() {
 		this(DirectProcessor.create());
 	}
 
-	public ReactiveChannel(FluxProcessor<Message<?>, Message<?>> processor) {
+	public FluxMessageChannel(FluxProcessor<Message<?>, Message<?>> processor) {
 		Assert.notNull(processor, "'processor' must not be null");
 		this.processor = processor;
-		this.flux = Flux.from(processor);
 		this.sink = processor.sink();
 	}
 
@@ -73,7 +73,7 @@ protected boolean doSend(Message<?> message, long timeout) {
 	public void subscribe(Subscriber<? super Message<?>> subscriber) {
 		this.subscribers.add(subscriber);
 
-		this.flux.doOnCancel(() -> ReactiveChannel.this.subscribers.remove(subscriber))
+		this.processor.doOnCancel(() -> FluxMessageChannel.this.subscribers.remove(subscriber))
 				.subscribe(subscriber);
 
 		if (!this.upstreamSubscribed) {
@@ -82,7 +82,7 @@ public void subscribe(Subscriber<? super Message<?>> subscriber) {
 	}
 
 	@Override
-	public void subscribeTo(Publisher<Message<?>> publisher) {
+	public void subscribeTo(Flux<Message<?>> publisher) {
 		this.publishers.add(publisher);
 		if (!this.subscribers.isEmpty()) {
 			doSubscribeTo(publisher);
@@ -91,11 +91,11 @@ public void subscribeTo(Publisher<Message<?>> publisher) {
 
 	private void doSubscribeTo(Publisher<Message<?>> publisher) {
 		Flux.from(publisher)
-				.doOnSubscribe(s -> ReactiveChannel.this.upstreamSubscribed = true)
+				.doOnSubscribe(s -> FluxMessageChannel.this.upstreamSubscribed = true)
 				.doOnComplete(() -> {
-					ReactiveChannel.this.publishers.remove(publisher);
-					if (ReactiveChannel.this.publishers.isEmpty()) {
-						ReactiveChannel.this.upstreamSubscribed = false;
+					FluxMessageChannel.this.publishers.remove(publisher);
+					if (FluxMessageChannel.this.publishers.isEmpty()) {
+						FluxMessageChannel.this.upstreamSubscribed = false;
 					}
 				})
 				.subscribe(this.processor);

spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveSubscribableChannel.java renamed to spring-integration-core/src/main/java/org/springframework/integration/channel/FluxSubscribableChannel.java

@@ -16,17 +16,18 @@
 
 package org.springframework.integration.channel;
 
-import org.reactivestreams.Publisher;
-
 import org.springframework.messaging.Message;
 
+import reactor.core.publisher.Flux;
+
 /**
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 5.0
  */
-public interface ReactiveSubscribableChannel {
+public interface FluxSubscribableChannel {
 
-	void subscribeTo(Publisher<Message<?>> publisher);
+	void subscribeTo(Flux<Message<?>> publisher);
 
 }

spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java

@@ -21,11 +21,11 @@
 
 import org.springframework.integration.dsl.channel.DirectChannelSpec;
 import org.springframework.integration.dsl.channel.ExecutorChannelSpec;
+import org.springframework.integration.dsl.channel.FluxMessageChannelSpec;
 import org.springframework.integration.dsl.channel.MessageChannels;
 import org.springframework.integration.dsl.channel.PriorityChannelSpec;
 import org.springframework.integration.dsl.channel.PublishSubscribeChannelSpec;
 import org.springframework.integration.dsl.channel.QueueChannelSpec;
-import org.springframework.integration.dsl.channel.ReactiveChannelSpec;
 import org.springframework.integration.dsl.channel.RendezvousChannelSpec;
 import org.springframework.integration.store.ChannelMessageStore;
 import org.springframework.integration.store.PriorityCapableChannelMessageStore;
@@ -132,20 +132,20 @@ public ExecutorChannelSpec executor(String id, Executor executor) {
 	}
 
 
-	public ReactiveChannelSpec reactive() {
-		return MessageChannels.reactive();
+	public FluxMessageChannelSpec flux() {
+		return MessageChannels.flux();
 	}
 
-	public ReactiveChannelSpec reactive(String id) {
-		return MessageChannels.reactive(id);
+	public FluxMessageChannelSpec flux(String id) {
+		return MessageChannels.flux(id);
 	}
 
-	public ReactiveChannelSpec reactive(FluxProcessor<Message<?>, Message<?>> processor) {
-		return MessageChannels.reactive(processor);
+	public FluxMessageChannelSpec flux(FluxProcessor<Message<?>, Message<?>> processor) {
+		return MessageChannels.flux(processor);
 	}
 
-	public ReactiveChannelSpec reactive(String id, FluxProcessor<Message<?>, Message<?>> processor) {
-		return MessageChannels.reactive(id, processor);
+	public FluxMessageChannelSpec flux(String id, FluxProcessor<Message<?>, Message<?>> processor) {
+		return MessageChannels.flux(id, processor);
 	}
 
 	Channels() {

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java

@@ -38,8 +38,8 @@
 import org.springframework.integration.channel.ChannelInterceptorAware;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.FixedSubscriberChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.channel.MessageChannelReactiveUtils;
-import org.springframework.integration.channel.ReactiveChannel;
 import org.springframework.integration.channel.interceptor.WireTap;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
@@ -2550,7 +2550,7 @@ public <T> Publisher<Message<T>> toReactivePublisher() {
 				publisher = MessageChannelReactiveUtils.toPublisher(channelForPublisher);
 			}
 			else {
-				MessageChannel reactiveChannel = new ReactiveChannel();
+				MessageChannel reactiveChannel = new FluxMessageChannel();
 				publisher = (Publisher<Message<T>>) reactiveChannel;
 				channel(reactiveChannel);
 			}

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java

@@ -21,7 +21,7 @@
 import org.reactivestreams.Publisher;
 
 import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.channel.ReactiveChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.channel.MessageChannelSpec;
 import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
@@ -35,10 +35,13 @@
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
+import reactor.core.publisher.Flux;
+
 /**
  * The central factory for fluent {@link IntegrationFlowBuilder} API.
  *
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 5.0
  *
@@ -299,15 +302,15 @@ protected void onInit() {
 	}
 
 	/**
-	 * Populate a {@link ReactiveChannel} to the {@link IntegrationFlowBuilder} chain
+	 * Populate a {@link FluxMessageChannel} to the {@link IntegrationFlowBuilder} chain
 	 * and subscribe it to the provided {@link Publisher}.
 	 * @param publisher the {@link Publisher} to subscribe to.
 	 * @return new {@link IntegrationFlowBuilder}.
 	 */
-	public static IntegrationFlowBuilder from(Publisher<Message<?>> publisher) {
-		ReactiveChannel reactiveChannel = new ReactiveChannel();
+	public static IntegrationFlowBuilder from(Flux<Message<?>> publisher) {
+		FluxMessageChannel reactiveChannel = new FluxMessageChannel();
 		reactiveChannel.subscribeTo(publisher);
-		return from((MessageChannel) reactiveChannel);
+		return from(reactiveChannel);
 	}
 
 	private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway,

spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java renamed to spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/FluxMessageChannelSpec.java

@@ -16,7 +16,7 @@
 
 package org.springframework.integration.dsl.channel;
 
-import org.springframework.integration.channel.ReactiveChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.messaging.Message;
 
 import reactor.core.publisher.FluxProcessor;
@@ -27,14 +27,14 @@
  *
  * @since 5.0
  */
-public class ReactiveChannelSpec extends MessageChannelSpec<ReactiveChannelSpec, ReactiveChannel> {
+public class FluxMessageChannelSpec extends MessageChannelSpec<FluxMessageChannelSpec, FluxMessageChannel> {
 
-	ReactiveChannelSpec() {
-		this.channel = new ReactiveChannel();
+	FluxMessageChannelSpec() {
+		this.channel = new FluxMessageChannel();
 	}
 
-	ReactiveChannelSpec(FluxProcessor<Message<?>, Message<?>> processor) {
-		this.channel = new ReactiveChannel(processor);
+	FluxMessageChannelSpec(FluxProcessor<Message<?>, Message<?>> processor) {
+		this.channel = new FluxMessageChannel(processor);
 	}
 
 }

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

@@ -126,20 +126,22 @@ public static <S extends PublishSubscribeChannelSpec<S>> PublishSubscribeChannel
 		return MessageChannels.<S>publishSubscribe(executor).id(id);
 	}
 
-	public static ReactiveChannelSpec reactive() {
-		return new ReactiveChannelSpec();
+	public static FluxMessageChannelSpec flux() {
+		return new FluxMessageChannelSpec();
 	}
 
-	public static ReactiveChannelSpec reactive(String id) {
-		return reactive().id(id);
+	public static FluxMessageChannelSpec flux(String id) {
+		return flux()
+				.id(id);
 	}
 
-	public static ReactiveChannelSpec reactive(FluxProcessor<Message<?>, Message<?>> processor) {
-		return new ReactiveChannelSpec(processor);
+	public static FluxMessageChannelSpec flux(String id, FluxProcessor<Message<?>, Message<?>> processor) {
+		return flux(processor)
+				.id(id);
 	}
 
-	public static ReactiveChannelSpec reactive(String id, FluxProcessor<Message<?>, Message<?>> processor) {
-		return reactive(processor).id(id);
+	public static FluxMessageChannelSpec flux(FluxProcessor<Message<?>, Message<?>> processor) {
+		return new FluxMessageChannelSpec(processor);
 	}
 
 	private MessageChannels() {

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java

@@ -24,7 +24,7 @@
 import org.reactivestreams.Publisher;
 
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
-import org.springframework.integration.channel.ReactiveSubscribableChannel;
+import org.springframework.integration.channel.FluxSubscribableChannel;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.routingslip.RoutingSlipRouteStrategy;
@@ -190,7 +190,7 @@ else if (reply instanceof AbstractIntegrationMessageBuilder) {
 		}
 
 		if (this.async && (reply instanceof ListenableFuture<?> || reply instanceof Publisher<?>)) {
-			if (reply instanceof ListenableFuture<?> || !(getOutputChannel() instanceof ReactiveSubscribableChannel)) {
+			if (reply instanceof ListenableFuture<?> || !(getOutputChannel() instanceof FluxSubscribableChannel)) {
 				ListenableFuture<?> future;
 				if (reply instanceof ListenableFuture<?>) {
 					future = (ListenableFuture<?>) reply;
@@ -235,7 +235,7 @@ public void onFailure(Throwable ex) {
 				});
 			}
 			else {
-				((ReactiveSubscribableChannel) getOutputChannel())
+				((FluxSubscribableChannel) getOutputChannel())
 						.subscribeTo(Flux.from((Publisher<?>) reply)
 								.map(result -> createOutputMessage(result, requestHeaders)));
 			}

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

@@ -36,9 +36,9 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.channel.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.channel.ReactiveChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -57,10 +57,10 @@
  */
 @RunWith(SpringRunner.class)
 @DirtiesContext
-public class ReactiveChannelTests {
+public class FluxMessageChannelTests {
 
 	@Autowired
-	private MessageChannel reactiveChannel;
+	private MessageChannel fluxMessageChannel;
 
 	@Autowired
 	private MessageChannel queueChannel;
@@ -69,11 +69,11 @@ public class ReactiveChannelTests {
 	private PollableChannel errorChannel;
 
 	@Test
-	public void testReactiveMessageChannel() throws InterruptedException {
+	public void testFluxMessageChannel() throws InterruptedException {
 		QueueChannel replyChannel = new QueueChannel();
 
 		for (int i = 0; i < 10; i++) {
-			this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build());
+			this.fluxMessageChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build());
 		}
 
 		for (int i = 0; i < 9; i++) {
@@ -116,11 +116,11 @@ public QueueChannel errorChannel() {
 		}
 
 		@Bean
-		public MessageChannel reactiveChannel() {
-			return new ReactiveChannel();
+		public MessageChannel fluxMessageChannel() {
+			return new FluxMessageChannel();
 		}
 
-		@ServiceActivator(inputChannel = "reactiveChannel")
+		@ServiceActivator(inputChannel = "fluxMessageChannel")
 		public String handle(int payload) {
 			if (payload == 5) {
 				throw new IllegalStateException("intentional");

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

@@ -47,7 +47,7 @@
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.channel.ReactiveChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.endpoint.ReactiveConsumer;
 import org.springframework.integration.handler.MethodInvokingMessageHandler;
@@ -67,7 +67,7 @@ public class ReactiveConsumerTests {
 
 	@Test
 	public void testReactiveConsumerReactiveChannel() throws InterruptedException {
-		ReactiveChannel testChannel = new ReactiveChannel(EmitterProcessor.create(false));
+		FluxMessageChannel testChannel = new FluxMessageChannel(EmitterProcessor.create(false));
 
 		List<Message<?>> result = new LinkedList<>();
 		CountDownLatch stopLatch = new CountDownLatch(2);
@@ -224,7 +224,7 @@ public void testReactiveConsumerPollableChannel() throws InterruptedException {
 
 	@Test
 	public void testReactiveConsumerViaConsumerEndpointFactoryBean() throws Exception {
-		ReactiveChannel testChannel = new ReactiveChannel();
+		FluxMessageChannel testChannel = new FluxMessageChannel();
 
 		List<Message<?>> result = new LinkedList<>();
 		CountDownLatch stopLatch = new CountDownLatch(3);