From 4c94974261ce16afec79e631efd656c3f948eccb Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 23 Sep 2019 11:28:41 -0400 Subject: [PATCH] Fix handling of native encoding/decoding for functions --- pom.xml | 2 +- .../function/FunctionConfiguration.java | 84 +++++++++---------- 2 files changed, 41 insertions(+), 45 deletions(-) diff --git a/pom.xml b/pom.xml index 4092c24de..e26e89c35 100644 --- a/pom.xml +++ b/pom.xml @@ -25,7 +25,7 @@ 1.8 Californium-SR11 2.1 - 3.0.0.M3 + 3.0.0.BUILD-SNAPSHOT true true true 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 83866ff7b..52312ec59 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 @@ -21,6 +21,7 @@ import java.lang.reflect.Method; import java.lang.reflect.Type; import java.time.Duration; import java.util.Iterator; +import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; @@ -43,6 +44,7 @@ import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.boot.autoconfigure.AutoConfigureBefore; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.function.context.FunctionProperties; import org.springframework.cloud.function.context.PollableSupplier; import org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.function.context.catalog.FunctionInspector; @@ -62,8 +64,6 @@ import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceConfiguration; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.cloud.stream.messaging.Source; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.ConfigurableApplicationContext; @@ -77,7 +77,6 @@ import org.springframework.core.env.Environment; import org.springframework.core.type.MethodMetadata; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.MessageChannelReactiveUtils; -import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlowBuilder; import org.springframework.integration.dsl.IntegrationFlows; @@ -121,7 +120,7 @@ public class FunctionConfiguration { return null; } - return new FunctionChannelBindingInitializer(functionCatalog, functionInspector, functionProperties, bindableProxyFactories, serviceProperties); + return new FunctionChannelBindingInitializer(functionCatalog, functionProperties, bindableProxyFactories, serviceProperties); } /* @@ -140,8 +139,11 @@ public class FunctionConfiguration { : new String[] {}; for (String functionDefinition : functionDefinitions) { - String contentType = serviceProperties.getBindingProperties("output").getContentType(); - FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(functionDefinition, contentType); + BindingProperties bindingProperties = serviceProperties.getBindingProperties("output"); + + FunctionInvocationWrapper functionWrapper = bindingProperties.getProducer() != null && bindingProperties.getProducer().isUseNativeEncoding() + ? functionCatalog.lookup(functionDefinition) + : functionCatalog.lookup(functionDefinition, bindingProperties.getContentType()); if (functionWrapper != null && functionWrapper.isSupplier()) { Publisher beginPublishingTrigger = this.setupBindingTrigger(context); @@ -231,8 +233,6 @@ public class FunctionConfiguration { private final FunctionCatalog functionCatalog; - private final FunctionInspector functionInspector; - private final StreamFunctionProperties functionProperties; private final BindableProxyFactory[] bindableProxyFactories; @@ -242,10 +242,9 @@ public class FunctionConfiguration { private GenericApplicationContext context; - FunctionChannelBindingInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector, - StreamFunctionProperties functionProperties, BindableProxyFactory[] bindableProxyFactories, BindingServiceProperties serviceProperties) { + FunctionChannelBindingInitializer(FunctionCatalog functionCatalog, StreamFunctionProperties functionProperties, + BindableProxyFactory[] bindableProxyFactories, BindingServiceProperties serviceProperties) { this.functionCatalog = functionCatalog; - this.functionInspector = functionInspector; this.functionProperties = functionProperties; this.bindableProxyFactories = bindableProxyFactories; this.serviceProperties = serviceProperties; @@ -262,40 +261,36 @@ public class FunctionConfiguration { } else { SubscribableChannel messageChannel = this.determineChannelToSubscribeTo(bindableProxyFactory); - if (messageChannel != null && function != null) { - this.bindOrComposeSimpleFunctions(((IntegrationObjectSupport) messageChannel).getComponentName(), - (SubscribableChannel) messageChannel, bindableProxyFactory, functionDefinition); + if (messageChannel != null) { + this.bindOrComposeSimpleFunctions(messageChannel, bindableProxyFactory, functionDefinition); } } } }); } - private String getFunctionDefinition(BindableProxyFactory bindableProxyFactory) { - return bindableProxyFactory instanceof BindableFunctionProxyFactory - ? ((BindableFunctionProxyFactory) bindableProxyFactory).getFunctionDefinition() - : functionProperties.getDefinition(); - } - @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.context = (GenericApplicationContext) applicationContext; } + private String getFunctionDefinition(BindableProxyFactory bindableProxyFactory) { + return bindableProxyFactory instanceof BindableFunctionProxyFactory + ? ((BindableFunctionProxyFactory) bindableProxyFactory).getFunctionDefinition() + : this.functionProperties.getDefinition(); + } + private SubscribableChannel determineChannelToSubscribeTo(BindableProxyFactory bindableProxyFactory) { SubscribableChannel messageChannel = null; if (bindableProxyFactory instanceof BindableFunctionProxyFactory) { String channelName = ((BindableFunctionProxyFactory) bindableProxyFactory).getInputName(0); - messageChannel = context.getBean(channelName, SubscribableChannel.class); + messageChannel = this.context.getBean(channelName, SubscribableChannel.class); } else { // could be "input" or "output" if subscribing to existing Source - if (context.containsBean(Sink.INPUT)) { - messageChannel = context.getBean(Sink.INPUT, SubscribableChannel.class); - } - else if (context.containsBean(Source.OUTPUT)) { - messageChannel = context.getBean(Source.OUTPUT, SubscribableChannel.class); - } + messageChannel = this.context.containsBean("input") + ? this.context.getBean("input", SubscribableChannel.class) + : this.context.getBean("output", SubscribableChannel.class); } return messageChannel; } @@ -313,14 +308,14 @@ public class FunctionConfiguration { .map(bindingName -> this.serviceProperties.getBindings().get(bindingName).getContentType()) .toArray(String[]::new); - FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition, outputContentTypes); + FunctionInvocationWrapper function = this.functionCatalog.lookup(functionDefinition, outputContentTypes); if (isMultipleInputOutput(bindableProxyFactory)) { this.assertSupportedSignatures(function.getFunctionType()); } Publisher[] inputPublishers = inputBindingNames.stream().map(inputBindingName -> { - SubscribableChannel inputChannel = context.getBean(inputBindingName, SubscribableChannel.class); + SubscribableChannel inputChannel = this.context.getBean(inputBindingName, SubscribableChannel.class); return this.enhancePublisher(MessageChannelReactiveUtils.toPublisher(inputChannel), inputBindingName); }).toArray(Publisher[]::new); @@ -329,32 +324,31 @@ public class FunctionConfiguration { if (resultPublishers instanceof Iterable) { Iterator outputBindingIter = outputBindingNames.iterator(); ((Iterable) resultPublishers).forEach(publisher -> { - MessageChannel outputChannel = context.getBean(outputBindingIter.next(), MessageChannel.class); + MessageChannel outputChannel = this.context.getBean(outputBindingIter.next(), MessageChannel.class); Flux.from((Publisher) publisher).doOnNext(message -> outputChannel.send((Message) message)).subscribe(); }); } else { outputBindingNames.stream().forEach(outputBindingName -> { - MessageChannel outputChannel = context.getBean(outputBindingName, MessageChannel.class); + MessageChannel outputChannel = this.context.getBean(outputBindingName, MessageChannel.class); Flux.from((Publisher) resultPublishers).doOnNext(message -> outputChannel.send((Message) message)).subscribe(); }); } } - // - private void bindOrComposeSimpleFunctions(String channelName, SubscribableChannel messageChannel, + /* + * Will either bind function to destination or compose it to the existing flow + */ + private void bindOrComposeSimpleFunctions(SubscribableChannel messageChannel, BindableProxyFactory bindableProxyFactory, String functionDefinition) { - //TODO there is something about moving channel interceptors in AMCB (not sure if it is still required) - String channelType = (String) ((DirectWithAttributesChannel) messageChannel).getAttribute("type"); - if (Source.OUTPUT.equals(channelType) && functionProperties.isComposeFrom()) { + BindingProperties properties = this.serviceProperties.getBindingProperties(((AbstractMessageChannel) messageChannel).getBeanName()); + FunctionInvocationWrapper function = this.functionCatalog.lookup(functionDefinition, properties.getContentType()); + + if (this.functionProperties.isComposeFrom()) { logger.info("Composing at the head of 'output' channel"); - BindingProperties properties = this.serviceProperties.getBindingProperties(channelName); - FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition, properties.getContentType()); this.composeSimpleFunctionToExistingFlow(function, messageChannel, bindableProxyFactory); } else { - BindingProperties properties = this.serviceProperties.getBindingProperties(channelName); - FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition, properties.getContentType()); this.bindSimpleFunctions(function, messageChannel, bindableProxyFactory); } } @@ -406,7 +400,7 @@ public class FunctionConfiguration { ? this.serviceProperties.getBindingProperties(outputChannelName).getProducer() : null; ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function, consumerProperties, producerProperties)); - handler.setBeanFactory(context); + handler.setBeanFactory(this.context); handler.afterPropertiesSet(); return handler; } @@ -515,9 +509,11 @@ public class FunctionConfiguration { @Override public Message apply(Message message) { -// Map headersMap = (Map) ReflectionUtils -// .getField(this.headersField, message.getHeaders()); - + if (message != null && consumerProperties != null) { + Map headersMap = (Map) ReflectionUtils + .getField(this.headersField, message.getHeaders()); + headersMap.put(FunctionProperties.SKIP_CONVERSION_HEADER, consumerProperties.isUseNativeDecoding()); + } Object result = function.apply(message); if (result instanceof Publisher) {