Polishing some missed renaming

Author: artembilan

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

@@ -32,6 +32,9 @@
 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
  *
@@ -46,8 +49,6 @@ public class FluxMessageChannel extends AbstractMessageChannel
 
 	private final FluxProcessor<Message<?>, Message<?>> processor;
 
-	private final Flux<Message<?>> flux;
-
 	private final FluxSink<Message<?>> sink;
 
 	private volatile boolean upstreamSubscribed;
@@ -59,7 +60,6 @@ public FluxMessageChannel() {
 	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(() -> FluxMessageChannel.this.subscribers.remove(subscriber))
+		this.processor.doOnCancel(() -> FluxMessageChannel.this.subscribers.remove(subscriber))
 				.subscribe(subscriber);
 
 		if (!this.upstreamSubscribed) {

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

@@ -132,20 +132,20 @@ public ExecutorChannelSpec executor(String id, Executor executor) {
 	}
 
 
-	public FluxMessageChannelSpec reactive() {
-		return MessageChannels.reactive();
+	public FluxMessageChannelSpec flux() {
+		return MessageChannels.flux();
 	}
 
-	public FluxMessageChannelSpec reactive(String id) {
-		return MessageChannels.reactive(id);
+	public FluxMessageChannelSpec flux(String id) {
+		return MessageChannels.flux(id);
 	}
 
-	public FluxMessageChannelSpec reactive(FluxProcessor<Message<?>, Message<?>> processor) {
-		return MessageChannels.reactive(processor);
+	public FluxMessageChannelSpec flux(FluxProcessor<Message<?>, Message<?>> processor) {
+		return MessageChannels.flux(processor);
 	}
 
-	public FluxMessageChannelSpec 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/channel/MessageChannels.java

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