diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 73e042f5b..8152f00b7 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -701,6 +701,69 @@ As you can see there is one additional argument that you can provide - `binderTy NOTE: For cases where `spring.cloud.stream.source` property is used or the binding was already created under different binder, the `binderType` argument will have no effect. +====== Using channel interceptors with StreamBridge + +Since `StreamBridge` uses a `MessageChannel` to establish the output binding, you can activate channel interceptors when sending data through `StreamBridge`. +It is up to the application to decide which channel interceptors to apply on `StreamBridge`. +Spring Cloud Stream does not inject all the channel interceptors detected into `StreamBridge` unless they are annoatated with `@GlobalChannelInterceptor(patterns = "*")`. + +Let us assume that you have the following two different `StreamBridge` bindings in the application. + +`streamBridge.send("foo-out-0", message);` + +and + +`streamBridge.send("bar-out-0", message);` + +Now, if you want a channel interceptor applied on both the `StreamBridge` bindings, then you can declare the following `GlobalChannelInterceptor` bean. + +``` +@Bean +@GlobalChannelInterceptor(patterns = "*") +public ChannelInterceptor customInterceptor() { + return new ChannelInterceptor() { + @Override + public Message preSend(Message message, MessageChannel channel) { + ... + } + }; +} +``` + +However, if you don't like the global approach above and want to have a dedicated interceptor for each binding, then you can do the following. + +``` +@Bean +@GlobalChannelInterceptor(patterns = "foo-*") +public ChannelInterceptor fooInterceptor() { + return new ChannelInterceptor() { + @Override + public Message preSend(Message message, MessageChannel channel) { + ... + } + }; +} +``` + +and + +``` +@Bean +@GlobalChannelInterceptor(patterns = "bar-*") +public ChannelInterceptor barInterceptor() { + return new ChannelInterceptor() { + @Override + public Message preSend(Message message, MessageChannel channel) { + ... + } + }; +} +``` + +You have the flexibility to make the patterns more strict or customized to your business needs. + +With this approach, the application gets the ability to decide which interceptors to inject in `StreamBridge` rather than applying all the available interceptors. + ===== Reactive Functions support diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 98c290563..882ce6cf7 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -17,7 +17,6 @@ package org.springframework.cloud.stream.function; import java.util.LinkedHashMap; -import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.function.Function; @@ -40,12 +39,11 @@ import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.config.GlobalChannelInterceptorProcessor; import org.springframework.integration.support.MessageBuilder; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.SubscribableChannel; -import org.springframework.messaging.support.ChannelInterceptor; -import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; import org.springframework.util.StringUtils; @@ -63,6 +61,7 @@ import org.springframework.util.StringUtils; * done through a declared function. * * @author Oleg Zhurakousky + * @author Soby Chacko * @since 3.0.3 * */ @@ -236,7 +235,7 @@ public final class StreamBridge implements SmartInitializingSingleton { SubscribableChannel messageChannel = this.channelCache.get(destinationName); if (messageChannel == null && this.applicationContext.containsBean(destinationName)) { messageChannel = this.applicationContext.getBean(destinationName, SubscribableChannel.class); - this.addInterceptors((AbstractMessageChannel) messageChannel); + this.addInterceptors((AbstractMessageChannel) messageChannel, destinationName); } if (messageChannel == null) { messageChannel = new DirectWithAttributesChannel(); @@ -255,20 +254,15 @@ public final class StreamBridge implements SmartInitializingSingleton { this.bindingService.bindProducer(messageChannel, destinationName, false, binder); this.channelCache.put(destinationName, messageChannel); - this.addInterceptors((AbstractMessageChannel) messageChannel); + this.addInterceptors((AbstractMessageChannel) messageChannel, destinationName); } return messageChannel; } - private void addInterceptors(AbstractMessageChannel messageChannel) { - String[] interceptorNames = this.applicationContext.getBeanNamesForType(ChannelInterceptor.class); - List interceptors = messageChannel.getInterceptors(); - for (String interceptorName : interceptorNames) { - ChannelInterceptor interceptor = this.applicationContext.getBean(interceptorName, ChannelInterceptor.class); - if (!CollectionUtils.containsInstance(interceptors, interceptor)) { - messageChannel.addInterceptor(interceptor); - } - } + private void addInterceptors(AbstractMessageChannel messageChannel, String destinationName) { + final GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor = + this.applicationContext.getBean(GlobalChannelInterceptorProcessor.class); + globalChannelInterceptorProcessor.postProcessAfterInitialization(messageChannel, destinationName); } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java index b04215218..542eeaead 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * Copyright 2020-2021 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. @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.function; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -37,10 +38,14 @@ import org.springframework.cloud.stream.binding.BinderAwareChannelResolver.NewDe import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.config.GlobalChannelInterceptor; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.handler.LoggingHandler; 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.MessageBuilder; @@ -50,6 +55,7 @@ import static org.junit.Assert.fail; /** * * @author Oleg Zhurakousky + * @author Soby Chacko * */ @SuppressWarnings("deprecation") @@ -82,14 +88,12 @@ public class StreamBridgeTests { } @Test - public void testWithInterceptor() { + public void testWithInterceptorsMatchedAgainstAllPatterns() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration .getCompleteConfiguration(ConsumerConfiguration.class, InterceptorConfiguration.class)) .web(WebApplicationType.NONE).run( "--spring.cloud.function.definition=function", "--spring.jmx.enabled=false")) { - - StreamBridge bridge = context.getBean(StreamBridge.class); bridge.send("function-in-0", "hello foo"); @@ -100,6 +104,26 @@ public class StreamBridgeTests { } } + @Test + public void testWithInterceptorsRegisteredOnlyOnOutputChannel() throws InterruptedException { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration + .getCompleteConfiguration(GH2180Configuration.class)) + .web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false")) { + + MessageChannel inputChannel = context.getBean("inputChannel", MessageChannel.class); + inputChannel.send(MessageBuilder.withPayload("hello foo").build()); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + Message message = outputDestination.receive(100, "outgoing-out-0"); + assertThat(new String(message.getPayload())).isEqualTo("hello foo"); + assertThat(message.getHeaders().get("intercepted")).isEqualTo("true"); + //Ensure that the LoggingHandler in the first SI flow is invoked. + GH2180Configuration.LATCH1.await(10, TimeUnit.SECONDS); + //Ensure that the second SI flow does not trigger its LoggingHandler (aka wiretap/interceptor). + assertThat(GH2180Configuration.LATCH2.getCount()).isEqualTo(1); + } + } @Test public void testBindingPropertiesAreHonored() { @@ -325,6 +349,7 @@ public class StreamBridgeTests { @EnableAutoConfiguration public static class InterceptorConfiguration { @Bean + @GlobalChannelInterceptor(patterns = "*") public ChannelInterceptor interceptor() { return new ChannelInterceptor() { @Override @@ -399,4 +424,62 @@ public class StreamBridgeTests { .get(); } } + + @EnableAutoConfiguration + public static class GH2180Configuration { + + static CountDownLatch LATCH1 = new CountDownLatch(1); + static CountDownLatch LATCH2 = new CountDownLatch(1); + + @Bean + MessageChannel inputChannel() { + return new DirectChannel(); + } + + @Bean + MessageChannel otherInputChannel() { + return new DirectChannel(); + } + + @Bean + public IntegrationFlow someFlow(MessageHandler sendMessage, MessageChannel inputChannel) { + return IntegrationFlows.from(inputChannel) + .log(LoggingHandler.Level.INFO, (m) -> { + LATCH1.countDown(); + return "Going through the first flow: " + m.getPayload(); + }) + .handle(sendMessage) + .get(); + } + + @Bean + public IntegrationFlow someOtherFlow(MessageHandler sendMessage) { + return IntegrationFlows.from(otherInputChannel()) + .log(LoggingHandler.Level.INFO, (m) -> { + LATCH2.countDown(); + return "Going through the second flow: " + m.getPayload(); + }) + .handle(sendMessage) + .get(); + } + + @Bean + @GlobalChannelInterceptor(patterns = "outgoing-*") + public ChannelInterceptor fooInterceptor() { + return new ChannelInterceptor() { + @Override + public Message preSend(Message message, MessageChannel channel) { + return MessageBuilder.fromMessage(message).setHeader("intercepted", "true").build(); + } + }; + } + + @Bean + public MessageHandler sendMessage(StreamBridge streamBridge) { + return message -> { + streamBridge.send("outgoing-out-0", message); + }; + } + } + }