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

Consolidate support for native encoding/decoding

Browse Source 
Switch to function snapshot
This commit is contained in:
Oleg Zhurakousky
2020-10-23 12:39:44 +02:00
parent 705665f638
commit f22c8f6ace
4 changed files with 121 additions and 33 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>Dysprosium-SR6</reactor.version>
<objenesis.version>2.1</objenesis.version>
<spring-cloud-function.version>3.0.10.RELEASE</spring-cloud-function.version>
<spring-cloud-function.version>3.0.11.BUILD-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>

20
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,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()

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

@@ -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<Message, Object> {
private static class FunctionWrapper implements Function<Message<byte[]>, 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<byte[]> message) {
if (message != null && consumerProperties != null) {
Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
.getField(this.headersField, message.getHeaders());

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

@@ -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<byte[]>("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<byte[]>("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<byte[]>("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<byte[]>("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<Object, String> imperative() {
return x -> {
return x.getClass().getSimpleName();
};
}
@Bean
public Function<Flux<Object>, Flux<String>> reactive() {
return flux -> flux.map(x -> {
return x.getClass().getSimpleName();
});
}
}
@EnableAutoConfiguration
public static class SingleFunctionConfiguration {