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

GH-1883 Fixed native encoding support for functions

Browse Source 
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
This commit is contained in:
Oleg Zhurakousky
2020-01-09 20:14:56 +01:00
parent 65940f8904
commit d7c2b7aed8
2 changed files with 21 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java
View File

@@ -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() {

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

@@ -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<byte[]> apply(Message<byte[]> message) {
public Object apply(Message<byte[]> message) {
if (message != null && consumerProperties != null) {
Map<String, Object> headersMap = (Map<String, Object>) 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<byte[]>) result;
return result;
}
}