From d998c199a34f70074f4db09d1b458597030838ec Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Wed, 13 Jan 2021 17:32:14 +0100
Subject: [PATCH] GH-2078 Add support for honoring channel interceptors in
 StreamBridge

Resolves #2078
---
 .../cloud/stream/function/StreamBridge.java   | 18 ++++++++++
 .../stream/function/StreamBridgeTests.java    | 35 +++++++++++++++++++
 2 files changed, 53 insertions(+)

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 fe4193e4e..cc221403e 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,6 +17,7 @@
 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;
@@ -37,10 +38,13 @@ import org.springframework.cloud.stream.binding.BindingService;
 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.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;
 
@@ -182,6 +186,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);
 		}
 		if (messageChannel == null) {
 			messageChannel = new DirectWithAttributesChannel();
@@ -194,7 +199,20 @@ public final class StreamBridge implements SmartInitializingSingleton {
 
 			this.bindingService.bindProducer(messageChannel, destinationName, false);
 			this.channelCache.put(destinationName, messageChannel);
+			this.addInterceptors((AbstractMessageChannel) messageChannel);
 		}
+
 		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);
+			}
+		}
+	}
 }
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 debca9553..1ecdefe18 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
@@ -37,6 +37,8 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.MessageBuilder;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -55,6 +57,26 @@ public class StreamBridgeTests {
 		System.clearProperty("spring.cloud.function.definition");
 	}
 
+	@Test
+	public void testWithInterceptor() {
+		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");
+
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+			Message<byte[]> message = outputDestination.receive(100, "function-out-0");
+			assertThat(new String(message.getPayload())).isEqualTo("hello foo");
+			assertThat(message.getHeaders().get("intercepted")).isEqualTo("true");
+		}
+	}
+
+
 	@Test
 	public void testBindingPropertiesAreHonored() {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
@@ -276,6 +298,19 @@ public class StreamBridgeTests {
 		}
 	}
 
+	@EnableAutoConfiguration
+	public static class InterceptorConfiguration {
+		@Bean
+		public ChannelInterceptor interceptor() {
+			return new ChannelInterceptor() {
+				@Override
+				public Message<?> preSend(Message<?> message, MessageChannel channel) {
+					return MessageBuilder.fromMessage(message).setHeader("intercepted", "true").build();
+				}
+			};
+		}
+	}
+
 	@EnableAutoConfiguration
 	public static class TestConfiguration {