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

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.stream.function;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -32,6 +33,7 @@ import org.springframework.cloud.function.context.FunctionRegistry;
 import org.springframework.cloud.function.context.FunctionType;
 import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
 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;
@@ -48,6 +50,7 @@ import org.springframework.integration.support.MessageBuilder;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.StringUtils;
@@ -94,6 +97,8 @@ public final class StreamBridge implements SmartInitializingSingleton {
 
 	private final Map<String, FunctionInvocationWrapper> streamBridgeFunctionCache;
 
+	private FunctionInvocationHelper<?> functionInvocationHelper;
+
 	/**
 	 *
 	 * @param functionCatalog instance of {@link FunctionCatalog}
@@ -121,6 +126,7 @@ public final class StreamBridge implements SmartInitializingSingleton {
 				return remove;
 			}
 		};
+		this.functionInvocationHelper = applicationContext.getBean(FunctionInvocationHelper.class);
 		this.streamBridgeFunctionCache = new HashMap<>();
 	}
 
@@ -203,9 +209,6 @@ public final class StreamBridge implements SmartInitializingSingleton {
 	 */
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	public boolean send(String bindingName, @Nullable String binderName, Object data, MimeType outputContentType) {
-		if (!(data instanceof Message)) {
-			data = MessageBuilder.withPayload(data).build();
-		}
 		ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName);
 		MessageChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName);
 
@@ -214,12 +217,22 @@ public final class StreamBridge implements SmartInitializingSingleton {
 		if (producerProperties != null && producerProperties.isPartitioned()) {
 			functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties);
 		}
-		// this function is a pass through and is only required to force output conversion if necessary on SCF side.
-		if (data instanceof Message) {
-			data = MessageBuilder.fromMessage((Message) data).setHeader(MessageUtils.TARGET_PROTOCOL, "streamBridge").build();
+
+		String targetType = this.resolveBinderTargetType(bindingName, MessageChannel.class, this.applicationContext.getBean(BinderFactory.class));
+
+		Message<?> messageToSend = data instanceof Message
+				? 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);
 		}
-		Message<byte[]> resultMessage = (Message<byte[]>) functionToInvoke.apply(data);
+
+		resultMessage = (Message<?>) this.functionInvocationHelper.postProcessResult(resultMessage, null);
+
 		return messageChannel.send(resultMessage);
+
 	}
 
 	private synchronized FunctionInvocationWrapper getStreamBridgeFunction(String outputContentType, ProducerProperties producerProperties) {
@@ -286,6 +299,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

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.function;
 
 import java.lang.reflect.Field;
 import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -34,7 +35,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;
@@ -78,6 +81,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