From f22c8f6ace8608a46287bdc866acce3a68edf78d Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 23 Oct 2020 12:39:44 +0200 Subject: [PATCH] Consolidate support for native encoding/decoding Switch to function snapshot --- pom.xml | 2 +- .../ObjectStringMessageConverter.java | 20 ++++- .../function/FunctionConfiguration.java | 42 ++++----- .../ImplicitFunctionBindingTests.java | 90 +++++++++++++++++-- 4 files changed, 121 insertions(+), 33 deletions(-) diff --git a/pom.xml b/pom.xml index 86e7a10f2..749862dc5 100644 --- a/pom.xml +++ b/pom.xml @@ -25,7 +25,7 @@ 1.8 Dysprosium-SR6 2.1 - 3.0.10.RELEASE + 3.0.11.BUILD-SNAPSHOT true true true 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..2f7d4e86b 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,6 @@ 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"); if (message.getPayload() != null) { if (message.getPayload() instanceof byte[]) { if (byte[].class.isAssignableFrom(targetClass)) { @@ -82,6 +82,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 b89d47853..df913d4ad 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 @@ -423,6 +423,7 @@ public class FunctionConfiguration { BindingProperties bindingProperties = this.serviceProperties.getBindings().get(inputBindingName); ConsumerProperties consumerProperties = bindingProperties == null ? null : bindingProperties.getConsumer(); if (consumerProperties != null) { + function.setSkipInputConversion(consumerProperties.isUseNativeDecoding()); Assert.isTrue(consumerProperties.getConcurrency() <= 1, "Concurrency > 1 is not supported by reactive " + "consumer, given that project reactor maintains its own concurrency mechanism. Was '..." + inputBindingName + ".consumer.concurrency=" + consumerProperties.getConcurrency() + "'"); @@ -435,6 +436,9 @@ public class FunctionConfiguration { if (!CollectionUtils.isEmpty(outputBindingNames)) { BindingProperties bindingProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next()); ProducerProperties producerProperties = bindingProperties == null ? null : bindingProperties.getProducer(); + if (producerProperties != null) { + function.setSkipOutputConversion(producerProperties.isUseNativeEncoding()); + } functionToInvoke = new PartitionAwareFunctionWrapper(function, this.applicationContext, producerProperties); } @@ -457,9 +461,15 @@ public class FunctionConfiguration { if (logger.isInfoEnabled()) { logger.info("Output message is sent to '" + destinationName + "' destination"); } + if (!(message instanceof Message)) { + message = MessageBuilder.withPayload(message).build(); + } dynamicChannel.send((Message) message); } else { + if (!(message instanceof Message)) { + message = MessageBuilder.withPayload(message).build(); + } outputChannel.send((Message) message); } }); @@ -471,9 +481,6 @@ public class FunctionConfiguration { } else { String outputDestinationName = this.determineOutputDestinationName(0, bindableProxyFactory, functionType); - if (StringUtils.hasText(outputDestinationName)) { - this.adjustFunctionForNativeEncodingIfNecessary(outputDestinationName, function, 0); - } String inputDestinationName = inputBindingNames.iterator().next(); Object inputDestination = this.applicationContext.getBean(inputDestinationName); if (inputDestination != null && inputDestination instanceof SubscribableChannel) { @@ -486,25 +493,6 @@ public class FunctionConfiguration { } } - private void adjustFunctionForNativeEncodingIfNecessary(String outputDestinationName, FunctionInvocationWrapper function, int index) { - if (function.isConsumer()) { - return; - } - BindingProperties properties = this.serviceProperties.getBindingProperties(outputDestinationName); - if (properties.getProducer() != null && properties.getProducer().isUseNativeEncoding()) { - Field acceptedOutputMimeTypesField = ReflectionUtils - .findField(FunctionInvocationWrapper.class, "acceptedOutputMimeTypes", String[].class); - acceptedOutputMimeTypesField.setAccessible(true); - try { - String[] acceptedOutputMimeTypes = (String[]) acceptedOutputMimeTypesField.get(function); - acceptedOutputMimeTypes[index] = ""; - } - catch (Exception e) { - // ignore - } - } - } - private ServiceActivatingHandler createFunctionHandler(FunctionInvocationWrapper function, String inputChannelName, String outputChannelName) { ConsumerProperties consumerProperties = StringUtils.hasText(inputChannelName) @@ -621,7 +609,7 @@ public class FunctionConfiguration { * not attempt any conversion and sends a raw Message. */ @SuppressWarnings("rawtypes") - private static class FunctionWrapper implements Function { + private static class FunctionWrapper implements Function, Object> { private final Function function; private final ConsumerProperties consumerProperties; @@ -642,14 +630,20 @@ public class FunctionConfiguration { this.applicationContext = applicationContext; this.function = new PartitionAwareFunctionWrapper((FunctionInvocationWrapper) function, this.applicationContext, producerProperties); this.consumerProperties = consumerProperties; + if (this.consumerProperties != null) { + ((FunctionInvocationWrapper) function).setSkipInputConversion(this.consumerProperties.isUseNativeDecoding()); + } this.producerProperties = producerProperties; + if (this.producerProperties != null) { + ((FunctionInvocationWrapper) function).setSkipOutputConversion(this.producerProperties.isUseNativeEncoding()); + } this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers"); this.headersField.setAccessible(true); } @SuppressWarnings("unchecked") @Override - public Object apply(Message message) { + public Object apply(Message message) { if (message != null && consumerProperties != null) { Map headersMap = (Map) ReflectionUtils .getField(this.headersField, message.getHeaders()); 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 91046aad7..858003f9d 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 @@ -869,17 +869,77 @@ public class ImplicitFunctionBindingTests { } } + @SuppressWarnings("rawtypes") @Test - public void foo() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration - .getCompleteConfiguration(PojoFunctionConfiguration.class)) + public void testWithNativeEncodingImperative() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(SingleFunctionConfiguration2.class)) .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", - "--spring.cloud.function.definition=f1|f2")) { + "--spring.cloud.function.definition=imperative")) { InputDestination inputDestination = context.getBean(InputDestination.class); + inputDestination.send(new GenericMessage("hello".getBytes())); + OutputDestination outputDestination = context.getBean(OutputDestination.class); - inputDestination.send(MessageBuilder.withPayload("hello".getBytes()).build()); - assertThat(outputDestination.receive(1000)).isNotNull(); + + Message result = outputDestination.receive(2000); + assertThat(result.getPayload()).isInstanceOf(byte[].class); // check output type + assertThat(new String((byte[]) result.getPayload())).isEqualTo("String"); // check input type + } + + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(SingleFunctionConfiguration2.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=imperative", + "--spring.cloud.stream.bindings.imperative-in-0.consumer.useNativeDecoding=true", + "--spring.cloud.stream.bindings.imperative-out-0.producer.useNativeEncoding=true" + )) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + inputDestination.send(new GenericMessage("hello".getBytes())); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message result = outputDestination.receive(2000); + assertThat(result.getPayload()).isInstanceOf(String.class); // check output type + assertThat(result.getPayload()).isEqualTo("byte[]"); // check input type + } + } + + @SuppressWarnings("rawtypes") + @Test + public void testWithNativeEncodingReactive() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(SingleFunctionConfiguration2.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=reactive")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + inputDestination.send(new GenericMessage("hello".getBytes())); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message result = outputDestination.receive(2000); + assertThat(result.getPayload()).isInstanceOf(byte[].class); // check output type + assertThat(new String((byte[]) result.getPayload())).isEqualTo("String"); // check input type + } + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(SingleFunctionConfiguration2.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", + "--spring.cloud.function.definition=reactive", + "--spring.cloud.stream.bindings.reactive-in-0.consumer.useNativeDecoding=true", + "--spring.cloud.stream.bindings.reactive-out-0.producer.useNativeEncoding=true")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + inputDestination.send(new GenericMessage("hello".getBytes())); + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message result = outputDestination.receive(2000); + assertThat(result.getPayload()).isInstanceOf(String.class); // no output conversion to byte[] has happened. + assertThat(result.getPayload()).isEqualTo("byte[]"); } } @@ -954,6 +1014,24 @@ public class ImplicitFunctionBindingTests { } } + @EnableAutoConfiguration + public static class SingleFunctionConfiguration2 { + + @Bean + public Function imperative() { + return x -> { + return x.getClass().getSimpleName(); + }; + } + + @Bean + public Function, Flux> reactive() { + return flux -> flux.map(x -> { + return x.getClass().getSimpleName(); + }); + } + } + @EnableAutoConfiguration public static class SingleFunctionConfiguration {