SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

StreamBridge and ChannelInterceptors enhancements

Browse Source 
Currently, when `StreamBridge` creates an output binding, the framework
applies all the available channel interceptors on it by default. This
creates unwanted side effects and the application does not have a way
to control this. Change this behavior by only applying channel interceptors
annoatated with `@GlobalChannelInterceptor`. This way, the user can
provide patterns to decide which interceptors need to be applied on
specific `StreamBridge` bindings.

Adding tests to verify.

Adding docs.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2180
This commit is contained in:
Soby Chacko
2021-06-10 19:30:15 -04:00
committed by Oleg Zhurakousky
parent 71f1af3c51
commit 76c0a2488b
3 changed files with 158 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
docs/src/main/asciidoc/spring-cloud-stream.adoc
View File

@@ -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

22
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -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<ChannelInterceptor> 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);
}
}

91
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java
View File

@@ -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<byte[]> 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);
};
}
}
}