From d7c2b7aed8f63300ab3bc7631c34c79dbfe6381c Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 9 Jan 2020 20:14:56 +0100 Subject: [PATCH] GH-1883 Fixed native encoding support for functions Addressed native encoding issue similar to the way it was done for the Supplier in #1885 where functions that are configured with useNativeEncoding should be looked up with no content type provided ensuring no output conversion is performed. Resolves #1883 --- .../function/BindableFunctionProxyFactory.java | 10 +++++++--- .../stream/function/FunctionConfiguration.java | 18 ++++++++++++++---- 2 files changed, 21 insertions(+), 7 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java index 2d9b2631e..bb687efb7 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java @@ -98,9 +98,13 @@ class BindableFunctionProxyFactory extends BindableProxyFactory { } protected String getOutputName(int index) { - return CollectionUtils.isEmpty(this.getOutputs()) - ? null - : this.getOutputs().toArray(new String[0])[index]; + String outputName = null; + if (this.outputCount > 0) { + outputName = CollectionUtils.isEmpty(this.getOutputs()) + ? null + : this.getOutputs().toArray(new String[0])[index]; + } + return outputName; } protected boolean isMultiple() { 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 4eeed95e1..966094334 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 @@ -406,8 +406,18 @@ public class FunctionConfiguration { */ private void bindOrComposeSimpleFunctions(SubscribableChannel messageChannel, BindableProxyFactory bindableProxyFactory, String functionDefinition) { - BindingProperties properties = this.serviceProperties.getBindingProperties(((AbstractMessageChannel) messageChannel).getBeanName()); - FunctionInvocationWrapper function = this.functionCatalog.lookup(functionDefinition, properties.getContentType()); + String channelName = ((AbstractMessageChannel) messageChannel).getBeanName(); + // see https://github.com/spring-cloud/spring-cloud-stream/issues/1883 for details + if (bindableProxyFactory instanceof BindableFunctionProxyFactory) { + String outputName = ((BindableFunctionProxyFactory) bindableProxyFactory).getOutputName(0); + if (StringUtils.hasText(outputName)) { + channelName = outputName; + } + } + BindingProperties properties = this.serviceProperties.getBindingProperties(channelName); + FunctionInvocationWrapper function = (properties.getProducer() != null && properties.getProducer().isUseNativeEncoding()) + ? this.functionCatalog.lookup(functionDefinition) + : this.functionCatalog.lookup(functionDefinition, properties.getContentType()); if (this.functionProperties.isComposeFrom()) { logger.info("Composing at the head of 'output' channel"); @@ -591,7 +601,7 @@ public class FunctionConfiguration { } @SuppressWarnings("unchecked") @Override - public Message apply(Message message) { + public Object apply(Message message) { if (message != null && consumerProperties != null) { Map headersMap = (Map) ReflectionUtils @@ -604,7 +614,7 @@ public class FunctionConfiguration { throw new IllegalStateException("Routing to functions that return Publisher is not supported " + "in the context of Spring Cloud Stream."); } - return (Message) result; + return result; } }