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

Fix FunctionConfiguration to account for Cloud Events recognition

Browse Source
This commit is contained in:
Oleg Zhurakousky
2020-12-21 13:03:20 +01:00
parent 263946265f
commit cedacc8b17
3 changed files with 34 additions and 16 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>2020.0.0-RC2</reactor.version>
<objenesis.version>2.1</objenesis.version>
<spring-cloud-function.version>3.1.0-RC1</spring-cloud-function.version>
<spring-cloud-function.version>3.1.0-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>

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java
View File

@@ -110,7 +110,12 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
StandardCharsets.UTF_8);
}
else {
result = super.convertFromInternal(message, targetClass, conversionHint);
try {
result = super.convertFromInternal(message, targetClass, conversionHint);
}
catch (Exception e) {
// ignore
}
}
}

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

@@ -53,6 +53,7 @@ import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.function.cloudevent.CloudEventMessageUtils;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.FunctionProperties;
import org.springframework.cloud.function.context.FunctionRegistry;
@@ -90,16 +91,18 @@ import org.springframework.core.env.Environment;
import org.springframework.core.type.MethodMetadata;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.AbstractSubscribableChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.IntegrationReactiveUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
@@ -445,6 +448,9 @@ public class FunctionConfiguration {
Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
.getField(headersField, ((Message) message).getHeaders());
headersMap.putIfAbsent(MessageUtils.TARGET_PROTOCOL, targetProtocol);
if (CloudEventMessageUtils.isCloudEvent((message))) {
headersMap.putIfAbsent(MessageUtils.MESSAGE_TYPE, CloudEventMessageUtils.CLOUDEVENT_VALUE);
}
return message;
});
}
@@ -527,16 +533,13 @@ public class FunctionConfiguration {
String inputDestinationName = inputBindingNames.iterator().next();
Object inputDestination = this.applicationContext.getBean(inputDestinationName);
if (inputDestination != null && inputDestination instanceof SubscribableChannel) {
ServiceActivatingHandler handler = createFunctionHandler(function, inputDestinationName, outputDestinationName);
if (StringUtils.hasText(outputDestinationName)) { // consumer implicit or function<.., mono<void>>
handler.setOutputChannelName(outputDestinationName);
}
AbstractMessageHandler handler = createFunctionHandler(function, inputDestinationName, outputDestinationName);
((SubscribableChannel) inputDestination).subscribe(handler);
}
}
}
private ServiceActivatingHandler createFunctionHandler(FunctionInvocationWrapper function,
private AbstractMessageHandler createFunctionHandler(FunctionInvocationWrapper function,
String inputChannelName, String outputChannelName) {
ConsumerProperties consumerProperties = StringUtils.hasText(inputChannelName)
? this.serviceProperties.getBindingProperties(inputChannelName).getConsumer()
@@ -545,22 +548,27 @@ public class FunctionConfiguration {
? this.serviceProperties.getBindingProperties(outputChannelName).getProducer()
: null;
ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function, consumerProperties,
producerProperties, applicationContext, this.determineTargetProtocol(outputChannelName))) {
FunctionWrapper functionInvocationWrapper = (new FunctionWrapper(function, consumerProperties,
producerProperties, applicationContext, this.determineTargetProtocol(outputChannelName)));
MessagingTemplate template = new MessagingTemplate();
template.setBeanFactory(applicationContext.getBeanFactory());
AbstractMessageHandler handler = new AbstractMessageHandler() {
@Override
protected void sendOutputs(Object result, Message<?> requestMessage) {
public void handleMessageInternal(Message<?> message) throws MessagingException {
Object result = functionInvocationWrapper.apply((Message<byte[]>) message);
if (result instanceof Iterable) {
for (Object resultElement : (Iterable<?>) result) {
this.doSendMessage(resultElement, requestMessage);
this.doSendMessage(resultElement, message);
}
}
else if (ObjectUtils.isArray(result) && !(result instanceof byte[])) {
for (int i = 0; i < ((Object[]) result).length; i++) {
this.doSendMessage(((Object[]) result)[i], requestMessage);
this.doSendMessage(((Object[]) result)[i], message);
}
}
else {
this.doSendMessage(result, requestMessage);
this.doSendMessage(result, message);
}
}
@@ -573,11 +581,13 @@ public class FunctionConfiguration {
}
outputChannel.send(((Message<?>) result));
}
else {
super.sendOutputs(result, requestMessage);
else if (StringUtils.hasText(outputChannelName)) {
template.send(outputChannelName, (Message<?>) result);
}
}
};
handler.setBeanFactory(this.applicationContext);
handler.afterPropertiesSet();
return handler;
@@ -694,6 +704,9 @@ public class FunctionConfiguration {
if (StringUtils.hasText(targetProtocol)) {
headersMap.putIfAbsent(MessageUtils.TARGET_PROTOCOL, targetProtocol);
}
if (CloudEventMessageUtils.isCloudEvent(message)) {
headersMap.putIfAbsent(MessageUtils.MESSAGE_TYPE, CloudEventMessageUtils.CLOUDEVENT_VALUE);
}
if (message != null && consumerProperties != null) {
headersMap.put(FunctionProperties.SKIP_CONVERSION_HEADER, consumerProperties.isUseNativeDecoding());
}