From 1511dd8748dbb72077111144f8460d97fc5068d2 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 5 Mar 2020 16:15:52 -0500 Subject: [PATCH] GH-3204: Add DSL intercept() operator Fixes https://github.com/spring-projects/spring-integration/issues/3204 * Add an `intercept(ChannelInterceptor...)` method into `BaseIntegrationFlowDefinition` to register one or more channel interceptors at the current flow position. * refactor to reuse `InterceptableChannel` creation from `wireTap` * document the new operator --- .../dsl/BaseIntegrationFlowDefinition.java | 49 +++++++++++-- .../dsl/flows/IntegrationFlowTests.java | 71 +++++++++++++++++++ src/reference/asciidoc/dsl.adoc | 18 ++++- src/reference/asciidoc/whats-new.adoc | 3 + 4 files changed, 133 insertions(+), 8 deletions(-) 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 4a65fcfdad..aefd5bbc5f 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 @@ -90,6 +90,7 @@ import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.InterceptableChannel; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -109,6 +110,7 @@ import reactor.util.function.Tuple2; * @author Artem Bilan * @author Gary Russell * @author Gabriele Del Prete + * @author Tim Feuerbach * * @since 5.2.1 * @@ -179,6 +181,24 @@ public abstract class BaseIntegrationFlowDefinition outputStringList; + + @Test + public void testInterceptorFlow() { + this.interceptorChannelIn.send(MessageBuilder.withPayload("foo").build()); + + assertThat(outputStringList).containsExactly( + "Pre send transform: foo", + "Pre send handle: FOO", + "Handle: FOO", + "Post send handle: FOO", + "Post send transform: foo" + ); + } + + @After + public void cleanUpList() { + outputStringList.clear(); + } + + @MessagingGateway public interface ControlBusGateway { @@ -909,6 +940,46 @@ public class IntegrationFlowTests { } + @Configuration + public static class InterceptorContextConfiguration { + + @Bean + public List outputStringList() { + return new ArrayList<>(); + } + + @Bean + public IntegrationFlow interceptorFlow(List outputStringList) { + return IntegrationFlows.from("interceptorChannelIn") + .intercept(new ChannelInterceptor() { + @Override + public Message preSend(Message message, MessageChannel channel) { + outputStringList.add("Pre send transform: " + message.getPayload()); + return message; + } + + @Override + public void postSend(Message message, MessageChannel channel, boolean sent) { + outputStringList.add("Post send transform: " + message.getPayload()); + } + }) + .transform((String s) -> s.toUpperCase()) + .intercept(new ChannelInterceptor() { + @Override + public Message preSend(Message message, MessageChannel channel) { + outputStringList.add("Pre send handle: " + message.getPayload()); + return message; + } + + @Override + public void postSend(Message message, MessageChannel channel, boolean sent) { + outputStringList.add("Post send handle: " + message.getPayload()); + } + }) + .handle(m -> outputStringList.add("Handle: " + m.getPayload())).get(); + } + } + @Service public static class GreetingService extends AbstractReplyProducingMessageHandler { diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index cd91bd1f26..30ad66a57a 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -594,6 +594,22 @@ When this operator is used at the end of a flow, it is a one-way handler and the To make it as a reply-producing flow, you can either use a simple `bridge()` after the `log()` or, starting with version 5.1, you can use a `logAndReply()` operator instead. `logAndReply` can only be used at the end of a flow. +[[java-dsl-intercept]] +=== Operator intercept() + +Starting with version 5.3, the `intercept()` operator allows to register one or more `ChannelInterceptor` instances at the current `MessageChannel` in the flow. +This is an alternative to creating an explicit `MessageChannel` via the `MessageChannels` API. +The following example uses a `MessageSelectingInterceptor` to reject certain messages with an exception: + +==== +[source,java] +---- +.transform(...) +.intercept(new MessageSelectingInterceptor(m -> m.getPayload().isValid())) +.handle(...) +---- +==== + [[java-dsl-wiretap]] === `MessageChannelSpec.wireTap()` @@ -618,7 +634,7 @@ public IntegrationFlow loggingFlow() { [IMPORTANT] ==== -If the `MessageChannel` is an instance of `InterceptableChannel`, the `log()` or `wireTap()` operators are applied to the current `MessageChannel`. +If the `MessageChannel` is an instance of `InterceptableChannel`, the `log()`, `wireTap()` or `intercept()` operators are applied to the current `MessageChannel`. Otherwise, an intermediate `DirectChannel` is injected into the flow for the currently configured endpoint. In the following example, the `WireTap` interceptor is added to `myChannel` directly, because `DirectChannel` implements `InterceptableChannel`: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 66f5ddbc3f..a9eea7b3a2 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -66,6 +66,9 @@ Transactional support in Spring Integration now also includes options to configu See `TransactionInterceptorBuilder` for more information. See also <<./transactions.adoc#reactive-transactions,Reactive Transactions>>. +A new `intercept()` operator to register `ChannelInterceptor` instances without creating explicit channels was added into Java DSL. +See <<./dsl.adoc#java-dsl-intercept,Operator intercept()>> for more information. + [[x5.3-amqp]] === AMQP Changes