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

Add support for SCF-GH-711 source/target types determination

Browse Source
This commit is contained in:
Oleg Zhurakousky
2021-07-01 17:26:18 +02:00
parent 8cae6f13b2
commit 32a8b9eec8
2 changed files with 16 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -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<? super Flux<Message<Object>>, ? extends Publisher<Object>>) 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());

6
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.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<byte[]> resultMessage = (Message<byte[]>) functionToInvoke.apply(data);
return messageChannel.send(resultMessage);
}