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

Consolidate support for native encoding/decoding

Browse Source
This commit is contained in:
Oleg Zhurakousky
2020-10-23 12:39:44 +02:00
parent bf9aee8b74
commit 5b04295570
2 changed files with 132 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -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<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=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[]");
}
}
@EnableAutoConfiguration
public static class SupplierAndPojoConfiguration {
@Bean
@@ -941,6 +1016,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 {