From ca231763a35995c665dcde8b1f3199a7e1c7d564 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 26 Apr 2017 11:38:13 -0400 Subject: [PATCH] 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 --- ...veChannel.java => FluxMessageChannel.java} | 26 +++++++++---------- ...nnel.java => FluxSubscribableChannel.java} | 9 ++++--- .../integration/dsl/Channels.java | 18 ++++++------- .../dsl/IntegrationFlowDefinition.java | 4 +-- .../integration/dsl/IntegrationFlows.java | 13 ++++++---- ...lSpec.java => FluxMessageChannelSpec.java} | 12 ++++----- .../dsl/channel/MessageChannels.java | 18 +++++++------ .../AbstractMessageProducingHandler.java | 6 ++--- ...ests.java => FluxMessageChannelTests.java} | 16 ++++++------ .../reactive/ReactiveConsumerTests.java | 6 ++--- ...tpRequestExecutingMessageHandlerTests.java | 4 +-- 11 files changed, 69 insertions(+), 63 deletions(-) rename spring-integration-core/src/main/java/org/springframework/integration/channel/{ReactiveChannel.java => FluxMessageChannel.java} (74%) rename spring-integration-core/src/main/java/org/springframework/integration/channel/{ReactiveSubscribableChannel.java => FluxSubscribableChannel.java} (83%) rename spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/{ReactiveChannelSpec.java => FluxMessageChannelSpec.java} (69%) rename spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/{ReactiveChannelTests.java => FluxMessageChannelTests.java} (89%) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java similarity index 74% rename from spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java index 1d3d902891..99dd819a25 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java @@ -32,13 +32,16 @@ import reactor.core.publisher.FluxProcessor; 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>, ReactiveSubscribableChannel { +public class FluxMessageChannel extends AbstractMessageChannel + implements Publisher>, FluxSubscribableChannel { private final List>> subscribers = new ArrayList<>(); @@ -46,20 +49,17 @@ public class ReactiveChannel extends AbstractMessageChannel private final FluxProcessor, Message> processor; - private final Flux> flux; - private final FluxSink> sink; private volatile boolean upstreamSubscribed; - public ReactiveChannel() { + public FluxMessageChannel() { this(DirectProcessor.create()); } - public ReactiveChannel(FluxProcessor, Message> processor) { + public FluxMessageChannel(FluxProcessor, 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 @@ public class ReactiveChannel extends AbstractMessageChannel public void subscribe(Subscriber> 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 class ReactiveChannel extends AbstractMessageChannel } @Override - public void subscribeTo(Publisher> publisher) { + public void subscribeTo(Flux> publisher) { this.publishers.add(publisher); if (!this.subscribers.isEmpty()) { doSubscribeTo(publisher); @@ -91,11 +91,11 @@ public class ReactiveChannel extends AbstractMessageChannel private void doSubscribeTo(Publisher> 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); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveSubscribableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxSubscribableChannel.java similarity index 83% rename from spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveSubscribableChannel.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/FluxSubscribableChannel.java index 36352266b9..756666ea91 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveSubscribableChannel.java +++ b/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> publisher); + void subscribeTo(Flux> publisher); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java index a67d2af349..c635c177db 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java @@ -21,11 +21,11 @@ import java.util.concurrent.Executor; 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 class Channels { } - 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> processor) { - return MessageChannels.reactive(processor); + public FluxMessageChannelSpec flux(FluxProcessor, Message> processor) { + return MessageChannels.flux(processor); } - public ReactiveChannelSpec reactive(String id, FluxProcessor, Message> processor) { - return MessageChannels.reactive(id, processor); + public FluxMessageChannelSpec flux(String id, FluxProcessor, Message> processor) { + return MessageChannels.flux(id, processor); } Channels() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index e86c5954e9..69a5ca7598 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -38,8 +38,8 @@ import org.springframework.integration.aggregator.BarrierMessageHandler; 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 abstract class IntegrationFlowDefinition>) reactiveChannel; channel(reactiveChannel); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index 09e9a781a2..de2eefc411 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -21,7 +21,7 @@ import java.util.function.Consumer; 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.Message; 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 @@ public final class IntegrationFlows { } /** - * 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> publisher) { - ReactiveChannel reactiveChannel = new ReactiveChannel(); + public static IntegrationFlowBuilder from(Flux> publisher) { + FluxMessageChannel reactiveChannel = new FluxMessageChannel(); reactiveChannel.subscribeTo(publisher); - return from((MessageChannel) reactiveChannel); + return from(reactiveChannel); } private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway, diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/FluxMessageChannelSpec.java similarity index 69% rename from spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java rename to spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/FluxMessageChannelSpec.java index c347d6d04b..5f8c261525 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java +++ b/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 @@ import reactor.core.publisher.FluxProcessor; * * @since 5.0 */ -public class ReactiveChannelSpec extends MessageChannelSpec { +public class FluxMessageChannelSpec extends MessageChannelSpec { - ReactiveChannelSpec() { - this.channel = new ReactiveChannel(); + FluxMessageChannelSpec() { + this.channel = new FluxMessageChannel(); } - ReactiveChannelSpec(FluxProcessor, Message> processor) { - this.channel = new ReactiveChannel(processor); + FluxMessageChannelSpec(FluxProcessor, Message> processor) { + this.channel = new FluxMessageChannel(processor); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java index 48391dea4d..5465d55837 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java @@ -126,20 +126,22 @@ public final class MessageChannels { return MessageChannels.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> processor) { - return new ReactiveChannelSpec(processor); + public static FluxMessageChannelSpec flux(String id, FluxProcessor, Message> processor) { + return flux(processor) + .id(id); } - public static ReactiveChannelSpec reactive(String id, FluxProcessor, Message> processor) { - return reactive(processor).id(id); + public static FluxMessageChannelSpec flux(FluxProcessor, Message> processor) { + return new FluxMessageChannelSpec(processor); } private MessageChannels() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index 8f2466ad9e..401e63ae9f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -24,7 +24,7 @@ import java.util.concurrent.atomic.AtomicInteger; 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 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } 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 abstract class AbstractMessageProducingHandler extends AbstractMessageHan }); } else { - ((ReactiveSubscribableChannel) getOutputChannel()) + ((FluxSubscribableChannel) getOutputChannel()) .subscribeTo(Flux.from((Publisher) reply) .map(result -> createOutputMessage(result, requestHeaders))); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java similarity index 89% rename from spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java index eac9b4f7d3..1161166ee8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java @@ -36,9 +36,9 @@ import org.springframework.beans.factory.annotation.Autowired; 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 @@ import reactor.core.publisher.Flux; */ @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 class ReactiveChannelTests { } @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"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java index f6512b28f5..8a73265c49 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java @@ -47,7 +47,7 @@ import org.springframework.beans.factory.BeanFactory; 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> result = new LinkedList<>(); CountDownLatch stopLatch = new CountDownLatch(2); @@ -224,7 +224,7 @@ public class ReactiveConsumerTests { @Test public void testReactiveConsumerViaConsumerEndpointFactoryBean() throws Exception { - ReactiveChannel testChannel = new ReactiveChannel(); + FluxMessageChannel testChannel = new FluxMessageChannel(); List> result = new LinkedList<>(); CountDownLatch stopLatch = new CountDownLatch(3); diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java index dde40bbf77..c0cc5cf8ee 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/outbound/ReactiveHttpRequestExecutingMessageHandlerTests.java @@ -29,7 +29,7 @@ import org.junit.Test; import org.springframework.http.HttpStatus; import org.springframework.http.client.reactive.ClientHttpConnector; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.channel.ReactiveChannel; +import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.http.HttpHeaders; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; @@ -63,7 +63,7 @@ public class ReactiveHttpRequestExecutingMessageHandlerTests { ReactiveHttpRequestExecutingMessageHandler reactiveHandler = new ReactiveHttpRequestExecutingMessageHandler(destinationUri, webClient); - ReactiveChannel ackChannel = new ReactiveChannel(); + FluxMessageChannel ackChannel = new FluxMessageChannel(); reactiveHandler.setOutputChannel(ackChannel); reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build());