SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Ensure StreamBridge function is always wrapped

Browse Source
This commit is contained in:
Oleg Zhurakousky
2021-07-05 13:29:25 +02:00
parent 32a8b9eec8
commit 6be9c461a5
1 changed files with 3 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -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<Object, Object> 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);