GH-SCF_856 Fix propper Cloud Event header prefix in StreamBridge

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java

@@ -34,6 +34,7 @@ import org.springframework.cloud.function.context.FunctionRegistry;
 import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
 import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.PassThruFunction;
 import org.springframework.cloud.function.context.message.MessageUtils;
+import org.springframework.cloud.function.core.FunctionInvocationHelper;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.BinderFactory;
 import org.springframework.cloud.stream.binder.ProducerProperties;
@@ -98,6 +99,8 @@ public final class StreamBridge implements SmartInitializingSingleton {
 
 	private final Map<String, FunctionInvocationWrapper> streamBridgeFunctionCache;
 
+	private FunctionInvocationHelper<?> functionInvocationHelper;
+
 	/**
 	 *
 	 * @param functionCatalog instance of {@link FunctionCatalog}
@@ -125,6 +128,7 @@ public final class StreamBridge implements SmartInitializingSingleton {
 				return remove;
 			}
 		};
+		this.functionInvocationHelper = applicationContext.getBean(FunctionInvocationHelper.class);
 		this.streamBridgeFunctionCache = new HashMap<>();
 	}
 
@@ -208,8 +212,6 @@ public final class StreamBridge implements SmartInitializingSingleton {
 	@SuppressWarnings({ "unchecked"})
 	public boolean send(String bindingName, @Nullable String binderName, Object data, MimeType outputContentType) {
 
-
-
 		ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName);
 		MessageChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName);
 
@@ -219,16 +221,19 @@ public final class StreamBridge implements SmartInitializingSingleton {
 			functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties);
 		}
 
+		String targetType = this.resolveBinderTargetType(bindingName, MessageChannel.class, this.applicationContext.getBean(BinderFactory.class));
 
 		Message<?> messageToSend = data instanceof Message
-				? MessageBuilder.fromMessage((Message) data).setHeader(MessageUtils.TARGET_PROTOCOL, "streamBridge").build()
-						: new GenericMessage<>(data, Collections.singletonMap(MessageUtils.TARGET_PROTOCOL, "streamBridge"));
+				? MessageBuilder.fromMessage((Message) data).setHeaderIfAbsent(MessageUtils.TARGET_PROTOCOL, targetType).build()
+						: new GenericMessage<>(data, Collections.singletonMap(MessageUtils.TARGET_PROTOCOL, targetType));
 
 		Message<?> resultMessage;
 		synchronized (this) {
 			resultMessage = (Message<byte[]>) functionToInvoke.apply(messageToSend);
 		}
 
+		resultMessage = (Message<?>) this.functionInvocationHelper.postProcessResult(resultMessage, null);
+
 		return messageChannel.send(resultMessage);
 	}
 
@@ -300,6 +305,14 @@ public final class StreamBridge implements SmartInitializingSingleton {
 		return messageChannel;
 	}
 
+	private String resolveBinderTargetType(String channelName, Class<?> bindableType, BinderFactory binderFactory) {
+		String binderConfigurationName = this.bindingServiceProperties
+				.getBinder(channelName);
+		Binder binder = binderFactory.getBinder(binderConfigurationName, bindableType);
+		String targetProtocol = binder.getClass().getSimpleName().startsWith("Rabbit") ? "amqp" : "kafka";
+		return targetProtocol;
+	}
+
 	private void addInterceptors(AbstractMessageChannel messageChannel, String destinationName) {
 		final GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor =
 			this.applicationContext.getBean(GlobalChannelInterceptorProcessor.class);

core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.function;
 import java.lang.reflect.Field;
 import java.util.Map;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -37,7 +38,9 @@ import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.function.cloudevent.CloudEventMessageBuilder;
 import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
+import org.springframework.cloud.function.context.message.MessageUtils;
 import org.springframework.cloud.stream.binder.test.InputDestination;
 import org.springframework.cloud.stream.binder.test.OutputDestination;
 import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
@@ -81,6 +84,24 @@ public class StreamBridgeTests {
 		System.clearProperty("spring.cloud.function.definition");
 	}
 
+	@Test
+	void test_SCF_856() throws Exception {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(EmptyConfiguration.class))
+				.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) {
+			StreamBridge streamBridge = context.getBean(StreamBridge.class);
+			streamBridge.send("myBinding-out-0",
+					CloudEventMessageBuilder.withData("hello").setSource("my-source")
+							.setId(UUID.randomUUID().toString()).setSpecVersion("1.0").setType("myType")
+							.setHeader(MessageUtils.TARGET_PROTOCOL, "kafka").build(),
+					MimeTypeUtils.APPLICATION_JSON);
+			OutputDestination output = context.getBean(OutputDestination.class);
+			Message<byte[]> result = output.receive();
+			assertThat(result.getHeaders().get("ce_type")).isNotNull();
+			assertThat(result.getHeaders().get("ce_source")).isNotNull();
+		}
+	}
+
 	/*
 	 * This test must not result in exception stating "Partition key cannot be null"
 	 * See https://github.com/spring-cloud/spring-cloud-stream/issues/2249 for more details