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

Fix handling of native encoding/decoding for functions

Browse Source
This commit is contained in:
Oleg Zhurakousky
2019-09-23 11:28:41 -04:00
parent 19f843ed18
commit 4c94974261
2 changed files with 41 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pom.xml
View File

@@ -25,7 +25,7 @@
<java.version>1.8</java.version>
<reactor.version>Californium-SR11</reactor.version>
<objenesis.version>2.1</objenesis.version>
<spring-cloud-function.version>3.0.0.M3</spring-cloud-function.version>
<spring-cloud-function.version>3.0.0.BUILD-SNAPSHOT</spring-cloud-function.version>
<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>
<maven-checkstyle-plugin.failsOnViolation>true</maven-checkstyle-plugin.failsOnViolation>
<maven-checkstyle-plugin.includeTestSourceDirectory>true</maven-checkstyle-plugin.includeTestSourceDirectory>

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

@@ -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<Object> 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<String> 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<byte[]> apply(Message<byte[]> message) {
// Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
// .getField(this.headersField, message.getHeaders());
if (message != null && consumerProperties != null) {
Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
.getField(this.headersField, message.getHeaders());
headersMap.put(FunctionProperties.SKIP_CONVERSION_HEADER, consumerProperties.isUseNativeDecoding());
}
Object result = function.apply(message);
if (result instanceof Publisher) {