From 77ee8ef1d3c268b661a0f69cdb6e96128f3cae85 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 26 Apr 2022 09:27:18 +0200 Subject: [PATCH] GH-SCF_856 Fix propper Cloud Event header prefix in StreamBridge --- .../cloud/stream/function/StreamBridge.java | 21 +++++++++++++++---- .../stream/function/StreamBridgeTests.java | 21 +++++++++++++++++++ 2 files changed, 38 insertions(+), 4 deletions(-) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index e9af62520..3ac90458c 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/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 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) 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); diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java index 63efe4307..50f88b36d 100644 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java +++ b/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 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