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

Cleanup and refactor to accomodate changes in SimpleFunctionRegistry of SCF"

Browse Source
This commit is contained in:
Oleg Zhurakousky
2020-10-20 15:07:47 +02:00
parent 14f1d285e4
commit 0c0aaec148
5 changed files with 83 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -72,6 +72,9 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
@Override
protected Object convertFromInternal(Message<?> message, Class<?> targetClass, @Nullable Object hint) {
if (message.getPayload().getClass().getName().startsWith("org.springframework.kafka.support.KafkaNull")) {
return null;
}
Object conversionHint = hint;
Object result = null;
if (conversionHint instanceof MethodParameter) {

22
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java
View File

@@ -18,12 +18,13 @@ package org.springframework.cloud.stream.converter;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.stream.Collectors;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.AbstractMessageConverter;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
/**
@@ -71,7 +72,8 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter {
@Override
protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) {
Assert.isTrue(String.class.isAssignableFrom(targetClass) || targetClass == Object.class, "This converter can only convert byte[] to String");
// Assert.isTrue(String.class.isAssignableFrom(targetClass) || targetClass == Object.class, "This converter can only convert byte[] to String");
if (message.getPayload() != null) {
if (message.getPayload() instanceof byte[]) {
if (byte[].class.isAssignableFrom(targetClass)) {
@@ -82,6 +84,22 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter {
StandardCharsets.UTF_8);
}
}
else if (message.getPayload() instanceof Collection) {
Collection<?> collection = ((Collection<?>) message.getPayload()).stream()
.map(value -> {
if (byte[].class.isAssignableFrom(targetClass)) {
return value;
}
else if (value instanceof byte[]) {
return new String((byte[]) value, StandardCharsets.UTF_8);
}
else {
return value; // String
}
}).collect(Collectors.toList());
return collection;
}
else {
if (byte[].class.isAssignableFrom(targetClass)) {
return message.getPayload().toString()

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

@@ -17,6 +17,7 @@
package org.springframework.cloud.stream.function;
import java.lang.reflect.Field;
import java.lang.reflect.GenericArrayType;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
@@ -200,7 +201,6 @@ public class FunctionConfiguration {
if (message.getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
String destinationName = (String) message.getHeaders().get("spring.cloud.stream.sendto.destination");
return streamBridge.resolveDestination(destinationName, producerProperties);
//return dynamicDestinationResolver.resolveDestination(destinationName);
}
return outputName;
}).get();
@@ -242,7 +242,7 @@ public class FunctionConfiguration {
boolean splittable = pollable != null
&& (boolean) AnnotationUtils.getAnnotationAttributes(pollable).get("splittable");
boolean reactive = FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0));
boolean reactive = FunctionTypeUtils.isPublisher(FunctionTypeUtils.getOutputType(functionType));
if (pollable == null && reactive) {
Publisher publisher = (Publisher) supplier.get();
@@ -366,7 +366,7 @@ public class FunctionConfiguration {
FunctionInvocationWrapper function = this.functionCatalog.lookup(functionDefinition, outputContentTypes);
Type functionType = function.getFunctionType();
this.assertSupportedSignatures(bindableProxyFactory, functionType);
this.assertSupportedSignatures(bindableProxyFactory, function);
if (this.functionProperties.isComposeFrom()) {
@@ -456,7 +456,7 @@ public class FunctionConfiguration {
BindingProperties properties = this.serviceProperties.getBindingProperties(outputDestinationName);
if (properties.getProducer() != null && properties.getProducer().isUseNativeEncoding()) {
Field acceptedOutputMimeTypesField = ReflectionUtils
.findField(FunctionInvocationWrapper.class, "acceptedOutputMimeTypes", String[].class);
.findField(FunctionInvocationWrapper.class, "expectedOutputContentType", String[].class);
acceptedOutputMimeTypesField.setAccessible(true);
try {
String[] acceptedOutputMimeTypes = (String[]) acceptedOutputMimeTypesField.get(function);
@@ -517,8 +517,8 @@ public class FunctionConfiguration {
private boolean isReactiveOrMultipleInputOutput(BindableProxyFactory bindableProxyFactory, Type functionType) {
boolean reactiveInputsOutputs = FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0)) ||
FunctionTypeUtils.isReactive(FunctionTypeUtils.getOutputType(functionType, 0));
boolean reactiveInputsOutputs = FunctionTypeUtils.isPublisher(FunctionTypeUtils.getInputType(functionType)) ||
FunctionTypeUtils.isPublisher(FunctionTypeUtils.getOutputType(functionType));
return isMultipleInputOutput(bindableProxyFactory) || reactiveInputsOutputs;
}
@@ -545,34 +545,38 @@ public class FunctionConfiguration {
&& ((BindableFunctionProxyFactory) bindableProxyFactory).isMultiple();
}
private void assertSupportedSignatures(BindableProxyFactory bindableProxyFactory, Type functionType) {
private boolean isArray(Type type) {
return type instanceof GenericArrayType || type instanceof Class && ((Class<?>) type).isArray();
}
private void assertSupportedSignatures(BindableProxyFactory bindableProxyFactory, FunctionInvocationWrapper function) {
if (this.isMultipleInputOutput(bindableProxyFactory)) {
Assert.isTrue(!FunctionTypeUtils.isConsumer(functionType),
Assert.isTrue(!function.isConsumer(),
"Function '" + functionProperties.getDefinition() + "' is a Consumer which is not supported "
+ "for multi-in/out reactive streams. Only Functions are supported");
Assert.isTrue(!FunctionTypeUtils.isSupplier(functionType),
Assert.isTrue(!function.isSupplier(),
"Function '" + functionProperties.getDefinition() + "' is a Supplier which is not supported "
+ "for multi-in/out reactive streams. Only Functions are supported");
Assert.isTrue(!FunctionTypeUtils.isInputArray(functionType) && !FunctionTypeUtils.isOutputArray(functionType),
Assert.isTrue(!this.isArray(function.getInputType()) && !this.isArray(function.getOutputType()),
"Function '" + functionProperties.getDefinition() + "' has the following signature: ["
+ functionType + "]. Your input and/or outout lacks arity and therefore we "
+ function.getFunctionType() + "]. Your input and/or outout lacks arity and therefore we "
+ "can not determine how many input/output destinations are required in the context of "
+ "function input/output binding.");
int inputCount = FunctionTypeUtils.getInputCount(functionType);
for (int i = 0; i < inputCount; i++) {
Assert.isTrue(FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, i)),
"Function '" + functionProperties.getDefinition() + "' has the following signature: ["
+ functionType + "]. Non-reactive functions with multiple "
+ "inputs/outputs are not supported in the context of Spring Cloud Stream.");
}
int outputCount = FunctionTypeUtils.getOutputCount(functionType);
for (int i = 0; i < outputCount; i++) {
Assert.isTrue(FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, i)),
"Function '" + functionProperties.getDefinition() + "' has the following signature: ["
+ functionType + "]. Non-reactive functions with multiple "
+ "inputs/outputs are not supported in the context of Spring Cloud Stream.");
}
// int inputCount = FunctionTypeUtils.getInputCount(function.getFunctionType());
// for (int i = 0; i < inputCount; i++) {
// Assert.isTrue(function.isInputTypePublisher(),
// "Function '" + functionProperties.getDefinition() + "' has the following signature: ["
// + function.getFunctionType() + "]. Non-reactive functions with multiple "
// + "inputs/outputs are not supported in the context of Spring Cloud Stream.");
// }
// int outputCount = FunctionTypeUtils.getOutputCount(function.getFunctionType());
// for (int i = 0; i < outputCount; i++) {
// Assert.isTrue(function.isOutputTypePublisher(),
// "Function '" + functionProperties.getDefinition() + "' has the following signature: ["
// + function.getFunctionType() + "]. Non-reactive functions with multiple "
// + "inputs/outputs are not supported in the context of Spring Cloud Stream.");
// }
}
}
@@ -744,11 +748,11 @@ public class FunctionConfiguration {
int outputCount = FunctionTypeUtils.getOutputCount(functionType);
if (!isSupplier && functionType instanceof ParameterizedType) {
Type outputType = ((ParameterizedType) functionType).getActualTypeArguments()[1];
if (FunctionTypeUtils.isOfType(outputType, Mono.class) && outputType instanceof ParameterizedType
&& FunctionTypeUtils.isOfType(((ParameterizedType) outputType).getActualTypeArguments()[0], Void.class)) {
if (FunctionTypeUtils.isMono(outputType) && outputType instanceof ParameterizedType
&& FunctionTypeUtils.getRawType(((ParameterizedType) outputType).getActualTypeArguments()[0]).equals(Void.class)) {
outputCount = 0;
}
else if (FunctionTypeUtils.isOfType(outputType, Void.class)) {
else if (FunctionTypeUtils.getRawType(outputType).equals(Void.class)) {
outputCount = 0;
}
}

45
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java
View File

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.function;
import java.lang.reflect.Field;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -31,6 +32,7 @@ import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.ReflectionUtils;
/**
* This class is effectively a wrapper which is aware of the stream related partition information
@@ -43,20 +45,27 @@ class PartitionAwareFunctionWrapper implements Function<Object, Object>, Supplie
private final FunctionInvocationWrapper function;
private final Field enhancerFiled;
@SuppressWarnings("rawtypes")
private final Function<Message, Message> outputMessageEnricher;
private final Function<Object, Message> outputMessageEnricher;
@SuppressWarnings("unchecked")
PartitionAwareFunctionWrapper(FunctionInvocationWrapper function, ConfigurableApplicationContext context, ProducerProperties producerProperties) {
this.function = function;
this.enhancerFiled = ReflectionUtils.findField(FunctionInvocationWrapper.class, "enhancer");
this.enhancerFiled.setAccessible(true);
if (producerProperties != null && producerProperties.isPartitioned()) {
StandardEvaluationContext evaluationContext = ExpressionUtils.createStandardEvaluationContext(context.getBeanFactory());
PartitionHandler partitionHandler = new PartitionHandler(evaluationContext, producerProperties, context.getBeanFactory());
this.outputMessageEnricher = outputMessage -> {
int partitionId = partitionHandler.determinePartition(outputMessage);
this.outputMessageEnricher = output -> {
if (!(output instanceof Message)) {
output = MessageBuilder.withPayload(output).build();
}
int partitionId = partitionHandler.determinePartition((Message<?>) output);
return MessageBuilder
.fromMessage(outputMessage)
.fromMessage((Message<?>) output)
.setHeader(BinderHeaders.PARTITION_HEADER, partitionId).build();
};
}
@@ -67,24 +76,24 @@ class PartitionAwareFunctionWrapper implements Function<Object, Object>, Supplie
@Override
public Object apply(Object input) {
if (this.outputMessageEnricher == null) { // to avoid breaking change
return this.function.apply(input);
}
try {
return this.function.apply(input, this.outputMessageEnricher);
}
catch (NoSuchMethodError e) {
logger.warn("Versions of spring-cloud-function older then 3.0.2.RELEASE do not support generation of partition information. "
+ "Output message will not contain any partition header unless spring-cloud-function dependency is 3.0.2.RELEASE or higher.");
return this.function.apply(input);
}
this.setEnhancerIfNecessary();
return this.function.apply(input);
}
@Override
public Object get() {
if (this.outputMessageEnricher == null) { // to avoid breaking change
return this.function.get();
this.setEnhancerIfNecessary();
return this.function.get();
}
private void setEnhancerIfNecessary() {
try {
// if (this.outputMessageEnricher == null) {
this.enhancerFiled.set(this.function, this.outputMessageEnricher);
// }
}
catch (Exception e) {
logger.warn("Failed to set the enhancer", e);
}
return this.function.get(this.outputMessageEnricher);
}
}

1
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
View File

@@ -820,6 +820,7 @@ public class ImplicitFunctionBindingTests {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
inputDestination.send(MessageBuilder.withPayload("hello").build());
assertThat(outputDestination.receive(1000, "output")).isNotNull();
assertThat(outputDestination.receive(1000, "output")).isNull();
assertThat(outputDestination.receive(1000, "output")).isNull();