From 87c8e47a886c74e7e7622f8dbee22fd30c0a8a55 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 25 Feb 2020 13:01:38 -0500 Subject: [PATCH] GH-3192: pub-sub DSL for broker-backed channels (#3193) * GH-3192: pub-sub DSL for broker-backed channels Fixes https://github.com/spring-projects/spring-integration/issues/3192 * Introduce a `BroadcastCapableChannel` abstract to indicate those `SubscribableChannel` implementations which can provide a pub-sub functionality * Implement a `BroadcastCapableChannel` in broker-baked channels with pub-sub option * Introduce a `BaseIntegrationFlowDefinition.publishSubscribeChannel()` based on the `BroadcastCapableChannel` and `BroadcastPublishSubscribeSpec` to let to configure sub-flow subscribers in fluent manner * * Add some JavaDocs and document new feature * * Show the channel bean definition in the doc * Fix typo --- .../channel/PublishSubscribeAmqpChannel.java | 15 ++-- .../channel/BroadcastCapableChannel.java | 38 +++++++++ .../channel/PublishSubscribeChannel.java | 20 ++--- .../dsl/BaseIntegrationFlowDefinition.java | 20 +++++ .../dsl/BroadcastPublishSubscribeSpec.java | 81 +++++++++++++++++++ .../integration/dsl/PublishSubscribeSpec.java | 29 ++----- .../jms/SubscribableJmsChannel.java | 41 ++++++---- .../integration/jms/dsl/JmsTests.java | 29 +++++-- .../channel/SubscribableRedisChannel.java | 17 ++-- src/reference/asciidoc/dsl.adoc | 34 ++++++++ src/reference/asciidoc/whats-new.adoc | 4 + 11 files changed, 265 insertions(+), 63 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/BroadcastCapableChannel.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/dsl/BroadcastPublishSubscribeSpec.java diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java index 42b953f556..e7b3d1d067 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -25,22 +25,25 @@ import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.channel.BroadcastCapableChannel; import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.dispatcher.BroadcastingDispatcher; /** + * The {@link AbstractSubscribableAmqpChannel} extension for pub-sub semantics based on the {@link FanoutExchange}. + * * @author Mark Fisher * @author Gary Russell * @author Artem Bilan * * @since 2.1 */ -public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel { - - private volatile FanoutExchange exchange; +public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel implements BroadcastCapableChannel { private final Queue queue = new AnonymousQueue(); + private volatile FanoutExchange exchange; + private volatile Binding binding; /** @@ -53,6 +56,7 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel */ public PublishSubscribeAmqpChannel(String channelName, AbstractMessageListenerContainer container, AmqpTemplate amqpTemplate) { + super(channelName, container, amqpTemplate, true); } @@ -69,6 +73,7 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel */ public PublishSubscribeAmqpChannel(String channelName, AbstractMessageListenerContainer container, AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) { + super(channelName, container, amqpTemplate, true, outboundMapper, inboundMapper); } @@ -104,7 +109,7 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel @Override protected AbstractDispatcher createDispatcher() { BroadcastingDispatcher broadcastingDispatcher = new BroadcastingDispatcher(true); - broadcastingDispatcher.setBeanFactory(this.getBeanFactory()); + broadcastingDispatcher.setBeanFactory(getBeanFactory()); return broadcastingDispatcher; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/BroadcastCapableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/BroadcastCapableChannel.java new file mode 100644 index 0000000000..af9b85b4f8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/BroadcastCapableChannel.java @@ -0,0 +1,38 @@ +/* + * Copyright 2020 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.channel; + +import org.springframework.messaging.SubscribableChannel; + +/** + * A {@link SubscribableChannel} variant for implementations with broadcasting capabilities. + * + * @author Artem Bilan + * + * @since 5.3 + */ +public interface BroadcastCapableChannel extends SubscribableChannel { + + /** + * Return a state of this channel in regards of broadcasting capabilities. + * @return the state of this channel in regards of broadcasting capabilities. + */ + default boolean isBroadcast() { + return true; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index 2e3241d66c..a182ea3d3f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -35,7 +35,7 @@ import org.springframework.util.ErrorHandler; * @author Gary Russell * @author Artem Bilan */ -public class PublishSubscribeChannel extends AbstractExecutorChannel { +public class PublishSubscribeChannel extends AbstractExecutorChannel implements BroadcastCapableChannel { private ErrorHandler errorHandler; @@ -45,6 +45,14 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel { private int minSubscribers; + /** + * Create a PublishSubscribeChannel that will invoke the handlers in the + * message sender's thread. + */ + public PublishSubscribeChannel() { + this(null); + } + /** * Create a PublishSubscribeChannel that will use an {@link Executor} * to invoke the handlers. If this is null, each invocation will occur in @@ -56,14 +64,6 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel { this.dispatcher = new BroadcastingDispatcher(executor); } - /** - * Create a PublishSubscribeChannel that will invoke the handlers in the - * message sender's thread. - */ - public PublishSubscribeChannel() { - this(null); - } - @Override public String getComponentType() { @@ -135,7 +135,7 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel { @Override public final void onInit() { super.onInit(); - BeanFactory beanFactory = this.getBeanFactory(); + BeanFactory beanFactory = getBeanFactory(); BroadcastingDispatcher dispatcherToUse = getDispatcher(); if (this.executor != null) { Assert.state(dispatcherToUse.getHandlerCount() == 0, diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java index 2a1f643058..4a65fcfdad 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java @@ -35,6 +35,7 @@ import org.springframework.beans.factory.config.DestructionAwareBeanPostProcesso import org.springframework.expression.Expression; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.aggregator.AggregatingMessageHandler; +import org.springframework.integration.channel.BroadcastCapableChannel; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.channel.FluxMessageChannel; @@ -294,6 +295,25 @@ public abstract class BaseIntegrationFlowDefinition publishSubscribeChannelConfigurer) { + + Assert.notNull(publishSubscribeChannelConfigurer, "'publishSubscribeChannelConfigurer' must not be null"); + BroadcastPublishSubscribeSpec spec = new BroadcastPublishSubscribeSpec(broadcastCapableChannel); + publishSubscribeChannelConfigurer.accept(spec); + return addComponents(spec.getComponentsToRegister()) + .channel(broadcastCapableChannel); + } + /** * Populate the {@code Wire Tap} EI Pattern specific * {@link org.springframework.messaging.support.ChannelInterceptor} implementation diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BroadcastPublishSubscribeSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BroadcastPublishSubscribeSpec.java new file mode 100644 index 0000000000..eceff70477 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BroadcastPublishSubscribeSpec.java @@ -0,0 +1,81 @@ +/* + * Copyright 2020 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.dsl; + +import java.util.LinkedHashMap; +import java.util.Map; + +import org.springframework.integration.channel.BroadcastCapableChannel; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; + +/** + * An {@link IntegrationComponentSpec} for configuring sub-flow subscribers on the + * provided {@link BroadcastCapableChannel}. + * + * @author Artem Bilan + * @author Gary Russell + * + * @since 5.3 + */ +public class BroadcastPublishSubscribeSpec + extends IntegrationComponentSpec + implements ComponentsRegistration { + + private final Map subscriberFlows = new LinkedHashMap<>(); + + private int order; + + protected BroadcastPublishSubscribeSpec(BroadcastCapableChannel channel) { + Assert.state(channel.isBroadcast(), + () -> "the " + channel + + " must be in the 'broadcast' state for using from this 'BroadcastPublishSubscribeSpec'"); + this.target = channel; + } + + /** + * Configure a {@link IntegrationFlow} to configure as a subscriber + * for the current {@link BroadcastCapableChannel}. + * @param subFlow the {@link IntegrationFlow} to configure as a subscriber + * for the current {@link BroadcastCapableChannel}. + * @return the current spec + */ + public BroadcastPublishSubscribeSpec subscribe(IntegrationFlow subFlow) { + Assert.notNull(subFlow, "'subFlow' must not be null"); + + IntegrationFlowBuilder flowBuilder = + IntegrationFlows.from(this.target) + .bridge(consumer -> consumer.order(this.order++)); + + MessageChannel subFlowInput = subFlow.getInputChannel(); + + if (subFlowInput == null) { + subFlow.configure(flowBuilder); + } + else { + flowBuilder.channel(subFlowInput); + } + this.subscriberFlows.put(flowBuilder.get(), null); + return _this(); + } + + @Override + public Map getComponentsToRegister() { + return this.subscriberFlows; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeSpec.java index 21df90a866..dd4fdae88b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeSpec.java @@ -21,10 +21,11 @@ import java.util.Map; import java.util.concurrent.Executor; import org.springframework.lang.Nullable; -import org.springframework.messaging.MessageChannel; -import org.springframework.util.Assert; /** + * The {@link PublishSubscribeChannelSpec} extension to configure as a general flow callback for sub-flows + * as subscribers. + * * @author Artem Bilan * @author Gary Russell * @@ -32,15 +33,15 @@ import org.springframework.util.Assert; */ public class PublishSubscribeSpec extends PublishSubscribeChannelSpec { - private final Map subscriberFlows = new LinkedHashMap<>(); - - private int order; + private final BroadcastPublishSubscribeSpec delegate; protected PublishSubscribeSpec() { + this.delegate = new BroadcastPublishSubscribeSpec(this.channel); } protected PublishSubscribeSpec(@Nullable Executor executor) { super(executor); + this.delegate = new BroadcastPublishSubscribeSpec(this.channel); } @Override @@ -49,21 +50,7 @@ public class PublishSubscribeSpec extends PublishSubscribeChannelSpec consumer.order(this.order++)); - - MessageChannel subFlowInput = subFlow.getInputChannel(); - - if (subFlowInput == null) { - subFlow.configure(flowBuilder); - } - else { - flowBuilder.channel(subFlowInput); - } - this.subscriberFlows.put(flowBuilder.get(), null); + this.delegate.subscribe(subFlow); return _this(); } @@ -71,7 +58,7 @@ public class PublishSubscribeSpec extends PublishSubscribeChannelSpec getComponentsToRegister() { Map objects = new LinkedHashMap<>(); objects.putAll(super.getComponentsToRegister()); - objects.putAll(this.subscriberFlows); + objects.putAll(this.delegate.getComponentsToRegister()); return objects; } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java index 8d059f0859..e977ad38ea 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -23,6 +23,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.context.SmartLifecycle; import org.springframework.integration.MessageDispatchingException; +import org.springframework.integration.channel.BroadcastCapableChannel; import org.springframework.integration.context.IntegrationProperties; import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.dispatcher.BroadcastingDispatcher; @@ -37,10 +38,13 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; -import org.springframework.messaging.SubscribableChannel; import org.springframework.util.Assert; /** + * An {@link AbstractJmsChannel} implementation for message-driven subscriptions. + * Also implements a {@link BroadcastCapableChannel} to represent possible pub-sub semantics + * when configured against JMS topic. + * * @author Mark Fisher * @author Gary Russell * @author Artem Bilan @@ -48,7 +52,7 @@ import org.springframework.util.Assert; * @since 2.0 */ public class SubscribableJmsChannel extends AbstractJmsChannel - implements SubscribableChannel, SmartLifecycle { + implements BroadcastCapableChannel, SmartLifecycle { private final AbstractMessageListenerContainer container; @@ -87,17 +91,22 @@ public class SubscribableJmsChannel extends AbstractJmsChannel return this.dispatcher.removeHandler(handler); } + @Override + public boolean isBroadcast() { + return this.container.isPubSubDomain(); + } + @Override public void onInit() { if (this.initialized) { return; } super.onInit(); - boolean isPubSub = this.container.isPubSubDomain(); - this.configureDispatcher(isPubSub); - MessageListener listener = new DispatchingMessageListener( - this.getJmsTemplate(), this.dispatcher, - this, isPubSub, this.getMessageBuilderFactory()); + boolean isPubSub = isBroadcast(); + configureDispatcher(isPubSub); + MessageListener listener = + new DispatchingMessageListener(getJmsTemplate(), this.dispatcher, this, isPubSub, + getMessageBuilderFactory()); this.container.setMessageListener(listener); if (!this.container.isActive()) { this.container.afterPropertiesSet(); @@ -108,7 +117,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel private void configureDispatcher(boolean isPubSub) { if (isPubSub) { BroadcastingDispatcher broadcastingDispatcher = new BroadcastingDispatcher(true); - broadcastingDispatcher.setBeanFactory(this.getBeanFactory()); + broadcastingDispatcher.setBeanFactory(getBeanFactory()); this.dispatcher = broadcastingDispatcher; } else { @@ -118,8 +127,8 @@ public class SubscribableJmsChannel extends AbstractJmsChannel } if (this.maxSubscribers == null) { this.maxSubscribers = this.getIntegrationProperty(isPubSub ? - IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS : - IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, + IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS : + IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class); } this.dispatcher.setMaxSubscribers(this.maxSubscribers); @@ -194,6 +203,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel DispatchingMessageListener(JmsTemplate jmsTemplate, MessageDispatcher dispatcher, SubscribableJmsChannel channel, boolean isPubSub, MessageBuilderFactory messageBuilderFactory) { + this.jmsTemplate = jmsTemplate; this.dispatcher = dispatcher; this.channel = channel; @@ -212,8 +222,10 @@ public class SubscribableJmsChannel extends AbstractJmsChannel converted = converter.fromMessage(message); } if (converted != null) { - messageToSend = (converted instanceof Message) ? (Message) converted - : this.messageBuilderFactory.withPayload(converted).build(); + messageToSend = + converted instanceof Message + ? (Message) converted + : this.messageBuilderFactory.withPayload(converted).build(); this.dispatcher.dispatch(messageToSend); } else if (this.logger.isWarnEnabled()) { @@ -231,8 +243,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel } } else { - throw new MessageDeliveryException( - messageToSend, exceptionMessage, e); + throw new MessageDeliveryException(messageToSend, exceptionMessage, e); } } catch (Exception e) { diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java index c088cc6d83..b5a656260c 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2020 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. @@ -41,6 +41,7 @@ import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.IntegrationComponentScan; import org.springframework.integration.annotation.MessagingGateway; import org.springframework.integration.annotation.Poller; +import org.springframework.integration.channel.BroadcastCapableChannel; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.channel.QueueChannel; @@ -125,6 +126,9 @@ public class JmsTests extends ActiveMQMultiContextTests { @Autowired private PollableChannel jmsPubSubBridgeChannel; + @Autowired + private PollableChannel jmsPubSubBridgeChannel2; + @Autowired @Qualifier("jmsOutboundGateway.handler") private MessageHandler jmsOutboundGatewayHandler; @@ -254,6 +258,11 @@ public class JmsTests extends ActiveMQMultiContextTests { .isNotNull() .extracting(Message::getPayload) .isEqualTo("foo"); + received = this.jmsPubSubBridgeChannel2.receive(5000); + assertThat(received) + .isNotNull() + .extracting(Message::getPayload) + .isEqualTo("foo"); } @Test @@ -351,10 +360,20 @@ public class JmsTests extends ActiveMQMultiContextTests { @Bean public IntegrationFlow pubSubFlow() { - return IntegrationFlows - .from(Jms.publishSubscribeChannel(jmsConnectionFactory()) - .destination("pubsub")) - .channel(c -> c.queue("jmsPubSubBridgeChannel")) + return f -> f + .publishSubscribeChannel(jmsPublishSubscribeChannel(), + pubsub -> pubsub + .subscribe(subFlow -> subFlow + .channel(c -> c.queue("jmsPubSubBridgeChannel"))) + .subscribe(subFlow -> subFlow + .channel(c -> c.queue("jmsPubSubBridgeChannel2")))); + } + + @Bean + public BroadcastCapableChannel jmsPublishSubscribeChannel() { + // TODO reconsider target generic type for channel implementation to return from this kind of specs + return (BroadcastCapableChannel) Jms.publishSubscribeChannel(jmsConnectionFactory()) + .destination("pubsub") .get(); } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java index 5a9bcde72c..ba08ef7a0f 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -32,6 +32,7 @@ import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.BroadcastCapableChannel; import org.springframework.integration.channel.ChannelUtils; import org.springframework.integration.context.IntegrationProperties; import org.springframework.integration.dispatcher.BroadcastingDispatcher; @@ -40,13 +41,15 @@ import org.springframework.integration.util.ErrorHandlingTaskExecutor; import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.converter.MessageConverter; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; import org.springframework.util.StringUtils; /** + * An {@link AbstractMessageChannel} implementation with {@link BroadcastCapableChannel} + * aspect to provide a pub-sub semantics to consume messages fgrom Redis topic. + * * @author Oleg Zhurakousky * @author Gary Russell * @author Artem Bilan @@ -55,7 +58,7 @@ import org.springframework.util.StringUtils; */ @SuppressWarnings("rawtypes") public class SubscribableRedisChannel extends AbstractMessageChannel - implements SubscribableChannel, SmartLifecycle { + implements BroadcastCapableChannel, SmartLifecycle { private final RedisMessageListenerContainer container = new RedisMessageListenerContainer(); @@ -67,10 +70,6 @@ public class SubscribableRedisChannel extends AbstractMessageChannel private final BroadcastingDispatcher dispatcher = new BroadcastingDispatcher(true); - private volatile Integer maxSubscribers; - - private volatile boolean initialized; - // defaults private Executor taskExecutor = new SimpleAsyncTaskExecutor(); @@ -78,6 +77,10 @@ public class SubscribableRedisChannel extends AbstractMessageChannel private MessageConverter messageConverter = new SimpleMessageConverter(); + private volatile Integer maxSubscribers; + + private volatile boolean initialized; + public SubscribableRedisChannel(RedisConnectionFactory connectionFactory, String topicName) { Assert.notNull(connectionFactory, "'connectionFactory' must not be null"); Assert.hasText(topicName, "'topicName' must not be empty"); diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 4efa05459e..cd91bd1f26 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -742,6 +742,40 @@ public IntegrationFlow subscribersFlow() { You can achieve the same result with separate `IntegrationFlow` `@Bean` definitions, but we hope you find the sub-flow style of logic composition useful. We find that it results in shorter (and so more readable) code. +Starting with version 5.3, a `BroadcastCapableChannel`-based `publishSubscribeChannel()` implementation is provided to configure sub-flow subscribers on broker-backed message channels. +For example we now can configure several subscribers as sub-flows on the `Jms.publishSubscribeChannel()`: + +==== +[source,java] +---- +@Bean +public BroadcastCapableChannel jmsPublishSubscribeChannel() { + return Jms.publishSubscribeChannel(jmsConnectionFactory()) + .destination("pubsub") + .get(); +} + +@Bean +public IntegrationFlow pubSubFlow() { + return f -> f + .publishSubscribeChannel(jmsPublishSubscribeChannel(), + pubsub -> pubsub + .subscribe(subFlow -> subFlow + .channel(c -> c.queue("jmsPubSubBridgeChannel1"))) + .subscribe(subFlow -> subFlow + .channel(c -> c.queue("jmsPubSubBridgeChannel2")))); +} + +@Bean +public BroadcastCapableChannel jmsPublishSubscribeChannel(ConnectionFactory jmsConnectionFactory) { + return (BroadcastCapableChannel) Jms.publishSubscribeChannel(jmsConnectionFactory) + .destination("pubsub") + .get(); +} + +---- +==== + A similar `publish-subscribe` sub-flow composition provides the `.routeToRecipients()` method. Another example is using `.discardFlow()` instead of `.discardChannel()` on the `.filter()` method. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 721a0371a4..05617fb369 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -53,6 +53,10 @@ See <<./graph.adoc#integration-graph,Integration Graph>> for more information. In the aggregator, when the `MessageGroupProcessor` returns a `Message`, the `MessageBuilder.popSequenceDetails()` is performed on the output message if the `sequenceDetails` matches the header in the first message of the group. See <<./aggregator.adoc#aggregator-api,Aggregator Programming Model>> for more information. +A new `publishSubscribeChannel()` operator, based on the `BroadcastCapableChannel` and `BroadcastPublishSubscribeSpec`, was added into Java DSL. +This fluent API has its advantage when we configure sub-flows as pub-sub subscribers for broker-backed channels like `SubscribableJmsChannel`, `SubscribableRedisChannel` etc. +See <<./dsl.adoc#java-dsl-subflows,Sub-flows support>> for more information. + [[x5.3-amqp]] === AMQP Changes