From 6be9c461a541a6683b21fd42500c1cbed4b08b3f Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 5 Jul 2021 13:29:25 +0200 Subject: [PATCH] Ensure StreamBridge function is always wrapped --- .../cloud/stream/function/StreamBridge.java | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) 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 76b36f38e..68831974f 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 @@ -29,6 +29,7 @@ import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.FunctionRegistration; 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.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderFactory; @@ -200,11 +201,8 @@ public final class StreamBridge implements SmartInitializingSingleton { ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); SubscribableChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName); - boolean skipConversion = producerProperties.isUseNativeEncoding(); - - Function functionToInvoke = skipConversion - ? (v -> v instanceof Message ? v : MessageBuilder.withPayload(v).build()) - : this.functionCatalog.lookup(STREAM_BRIDGE_FUNC_NAME, outputContentType.toString()); + Function functionToInvoke = this.functionCatalog.lookup(STREAM_BRIDGE_FUNC_NAME, outputContentType.toString()); + ((FunctionInvocationWrapper) functionToInvoke).setSkipOutputConversion(producerProperties.isUseNativeEncoding()); if (producerProperties != null && producerProperties.isPartitioned()) { functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties);