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

GH-1885, GH-1896 Added tests for conversion of collection of messages

Browse Source 
Also, fixed the routing exception when function returns a publisher

Resolves #1885
Resolves #1896
This commit is contained in:
Oleg Zhurakousky
2020-01-27 17:29:10 +01:00
parent 4077b0d728
commit 67b36959fd
2 changed files with 86 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -359,7 +359,7 @@ public class FunctionConfiguration {
this.adjustFunctionForNativeEncodingIfNecessary(outputDestinationName, function, 0);
if (this.functionProperties.isComposeFrom()) {
SubscribableChannel outputChannel = this.applicationContext.getBean(outputDestinationName, SubscribableChannel.class);
//logger.info("Composing at the head of 'output' channel");
// logger.info("Composing at the head of 'output' channel");
String outputChannelName = ((AbstractMessageChannel) outputChannel).getBeanName();
ServiceActivatingHandler handler = createFunctionHandler(function, null, outputChannelName);
@@ -510,6 +510,10 @@ public class FunctionConfiguration {
FunctionWrapper(Function function, ConsumerProperties consumerProperties, ProducerProperties producerProperties) {
this.function = function;
Type type = ((FunctionInvocationWrapper) function).getFunctionType();
if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getOutputType(type, 0))) {
throw new IllegalStateException("Functions with mixed semantics (imperative input vs. reactive output) ar not supported at the moment");
}
this.consumerProperties = consumerProperties;
this.producerProperties = producerProperties;
this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
@@ -518,17 +522,15 @@ public class FunctionConfiguration {
@SuppressWarnings("unchecked")
@Override
public Object apply(Message<byte[]> message) {
if (message != null && consumerProperties != null) {
Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
.getField(this.headersField, message.getHeaders());
headersMap.put(FunctionProperties.SKIP_CONVERSION_HEADER, consumerProperties.isUseNativeDecoding());
}
Object result = function.apply(message);
if (result instanceof Publisher) {
throw new IllegalStateException("Routing to functions that return Publisher is not supported "
+ "in the context of Spring Cloud Stream.");
if (result instanceof Publisher && ((FunctionInvocationWrapper) this.function).getTarget() instanceof RoutingFunction) {
throw new IllegalStateException("Routing to functions that return Publisher "
+ "is not supported in the context of Spring Cloud Stream.");
}
return result;
}

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

@@ -18,6 +18,8 @@ package org.springframework.cloud.stream.function;
import java.io.Serializable;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
@@ -394,6 +396,53 @@ public class ImplicitFunctionBindingTests {
}
}
// see https://github.com/spring-cloud/spring-cloud-stream/issues/1896
@Test
public void testOutputAsCollectionOfMessages() {
System.clearProperty("spring.cloud.function.definition");
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(SplittableTypesConfiguration.class))
.web(WebApplicationType.NONE).run("--spring.cloud.function.definition=funcCollectionOfMessages",
"--spring.jmx.enabled=false")) {
InputDestination inputDestination = context.getBean(InputDestination.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> inputMessage = MessageBuilder.withPayload("aa,bb,cc,dd".getBytes()).build();
inputDestination.send(inputMessage);
assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("aa");
assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("bb");
assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("cc");
assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("dd");
assertThat(outputDestination.receive(100)).isNull();
}
}
@Test
public void testOutputAsArrayOfMessages() {
System.clearProperty("spring.cloud.function.definition");
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(SplittableTypesConfiguration.class))
.web(WebApplicationType.NONE).run("--spring.cloud.function.definition=funcArrayOfMessages",
"--spring.jmx.enabled=false")) {
InputDestination inputDestination = context.getBean(InputDestination.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> inputMessage = MessageBuilder.withPayload("aa,bb,cc,dd".getBytes()).build();
inputDestination.send(inputMessage);
assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("aa");
assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("bb");
assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("cc");
assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("dd");
assertThat(outputDestination.receive(100)).isNull();
}
}
@EnableAutoConfiguration
public static class NoEnableBindingConfiguration {
@@ -541,15 +590,40 @@ public class ImplicitFunctionBindingTests {
@Bean
public Function<Message<List<Map<String, Integer>>>, Message<List<Map<String, Integer>>>> funcA() {
return v -> {
return v;
};
return v -> v;
}
@Bean
public Function<Message<List<Map<String, String>>>, Message<List<Map<String, String>>>> funcB() {
return v -> v;
}
}
@EnableAutoConfiguration
public static class SplittableTypesConfiguration {
@Bean
public Function<String, Collection<Message<String>>> funcCollectionOfMessages() {
return v -> {
return v;
String[] values = v.split(",");
List<Message<String>> messages = new ArrayList<>();
for (String value : values) {
messages.add(MessageBuilder.withPayload(value).build());
}
return messages;
};
}
@SuppressWarnings("unchecked")
@Bean
public Function<String, Message<String>[]> funcArrayOfMessages() {
return v -> {
String[] values = v.split(",");
List<Message<String>> messages = new ArrayList<>();
for (String value : values) {
messages.add(MessageBuilder.withPayload(value).build());
}
return messages.toArray(new Message[0]);
};
}
}