From 5b04295570f79d9bc9347c90f3412c2242febf5e 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 --- .../function/FunctionConfiguration.java | 62 ++++++++----- .../ImplicitFunctionBindingTests.java | 93 +++++++++++++++++++ 2 files changed, 132 insertions(+), 23 deletions(-) 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 88a8db50c..5ecf3e75b 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 @@ -432,6 +432,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() + "'"); @@ -444,6 +445,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); } @@ -466,9 +470,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); } }); @@ -480,9 +490,9 @@ public class FunctionConfiguration { } else { String outputDestinationName = this.determineOutputDestinationName(0, bindableProxyFactory, functionType); - if (StringUtils.hasText(outputDestinationName)) { - this.adjustFunctionForNativeEncodingIfNecessary(outputDestinationName, function, 0); - } +// 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) { @@ -495,24 +505,24 @@ 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, "expectedOutputContentType", String[].class); - acceptedOutputMimeTypesField.setAccessible(true); - try { - String[] acceptedOutputMimeTypes = (String[]) acceptedOutputMimeTypesField.get(function); - acceptedOutputMimeTypes[index] = ""; - } - catch (Exception e) { - // ignore - } - } - } +// 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, "expectedOutputContentType", 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) { @@ -634,7 +644,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; @@ -655,14 +665,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 072318fe5..8f5d28073 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 @@ -870,6 +870,81 @@ public class ImplicitFunctionBindingTests { } } + @SuppressWarnings("rawtypes") + @Test + 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=imperative")) { + + 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=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[]"); + } + } + + @EnableAutoConfiguration public static class SupplierAndPojoConfiguration { @Bean @@ -941,6 +1016,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 {