From 0c0aaec14851d7af5fa0af45784024e54b49e12e Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 20 Oct 2020 15:07:47 +0200 Subject: [PATCH] Cleanup and refactor to accomodate changes in SimpleFunctionRegistry of SCF" --- ...cationJsonMessageMarshallingConverter.java | 3 + .../ObjectStringMessageConverter.java | 22 ++++++- .../function/FunctionConfiguration.java | 60 ++++++++++--------- .../PartitionAwareFunctionWrapper.java | 45 ++++++++------ .../ImplicitFunctionBindingTests.java | 1 + 5 files changed, 83 insertions(+), 48 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java index b5d3ee1d4..18e560afd 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java @@ -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) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java index 76b61d4fd..bb1e55a1a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java @@ -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() 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 01e2c4387..45f1b844d 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 @@ -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; } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java index a0373cde3..0f9328d5c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java @@ -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, Supplie private final FunctionInvocationWrapper function; + private final Field enhancerFiled; + @SuppressWarnings("rawtypes") - private final Function outputMessageEnricher; + private final Function 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, 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); } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java index ba8284f77..4ca3fc0af 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java @@ -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();