From 32a8b9eec802a5c13cb7602089983ca1bb2b3a40 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 1 Jul 2021 17:26:18 +0200 Subject: [PATCH] Add support for SCF-GH-711 source/target types determination --- .../stream/function/FunctionConfiguration.java | 14 +++++++++++--- .../cloud/stream/function/StreamBridge.java | 6 +++++- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index 1a44e70f6..59346006e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -243,7 +243,7 @@ public class FunctionConfiguration { } else { Type functionType = ((FunctionInvocationWrapper) supplier).getFunctionType(); - IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper((FunctionInvocationWrapper) supplier, context, producerProperties), + IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper(supplier, context, producerProperties), beginPublishingTrigger, pollable, context, taskScheduler, functionType) .channel(c -> c.direct()) .fluxTransform((Function>, ? extends Publisher>) function) @@ -354,7 +354,15 @@ public class FunctionConfiguration { @SuppressWarnings({ "unchecked", "rawtypes" }) private static Message sanitize(Message inputMessage) { - return MessageBuilder.fromMessage(inputMessage).removeHeader("spring.cloud.stream.sendto.destination").build(); + MessageBuilder builder = MessageBuilder + .fromMessage(inputMessage) + .removeHeader("spring.cloud.stream.sendto.destination") + .removeHeader(MessageUtils.SOURCE_TYPE); + if (builder.getHeaders().containsKey(MessageUtils.TARGET_PROTOCOL)) { + builder = builder.setHeader(MessageUtils.SOURCE_TYPE, builder.getHeaders().get(MessageUtils.TARGET_PROTOCOL)); + } + builder = builder.removeHeader(MessageUtils.TARGET_PROTOCOL); + return builder.build(); } private static class FunctionToDestinationBinder implements InitializingBean, ApplicationContextAware { @@ -706,7 +714,7 @@ public class FunctionConfiguration { isRoutingFunction = ((FunctionInvocationWrapper) function).getTarget() instanceof RoutingFunction; this.applicationContext = applicationContext; - this.function = new PartitionAwareFunctionWrapper((FunctionInvocationWrapper) function, this.applicationContext, producerProperties); + this.function = new PartitionAwareFunctionWrapper(function, this.applicationContext, producerProperties); this.consumerProperties = consumerProperties; if (this.consumerProperties != null) { ((FunctionInvocationWrapper) function).setSkipInputConversion(this.consumerProperties.isUseNativeDecoding()); 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 0e4ef8e13..76b36f38e 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.message.MessageUtils; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.ProducerProperties; @@ -191,7 +192,7 @@ public final class StreamBridge implements SmartInitializingSingleton { * @param outputContentType content type to be used to deal with output type conversion * @return true if data was sent successfully, otherwise false or throws an exception. */ - @SuppressWarnings("unchecked") + @SuppressWarnings({ "unchecked", "rawtypes" }) public boolean send(String bindingName, @Nullable String binderName, Object data, MimeType outputContentType) { if (!(data instanceof Message)) { data = MessageBuilder.withPayload(data).build(); @@ -209,6 +210,9 @@ public final class StreamBridge implements SmartInitializingSingleton { 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(); + } Message resultMessage = (Message) functionToInvoke.apply(data); return messageChannel.send(resultMessage); }