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

GH-1834 Fix reactive consumer

Browse Source 
Added null check for reactive wirong in FunctionConfiguration

Resolves #1834
This commit is contained in:
Oleg Zhurakousky
2019-10-28 17:43:25 +01:00
parent 44f114e4e7
commit a054f047f3
2 changed files with 32 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -415,7 +415,7 @@ public class FunctionConfiguration {
? ((BindableFunctionProxyFactory) bindableProxyFactory).getOutputName(0)
: (FunctionTypeUtils.isConsumer(functionType) ? null : "output");
if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))) {
if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0)) && StringUtils.hasText(outputChannelName)) {
MessageChannel outputChannel = context.getBean(outputChannelName, MessageChannel.class);
SubscribableChannel subscribeChannel = (SubscribableChannel) inputChannel;
Publisher<?> publisher = this.enhancePublisher(MessageChannelReactiveUtils.toPublisher(subscribeChannel),

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

@@ -167,6 +167,25 @@ public class ImplicitFunctionBindingTests {
}
}
@Test
public void testReactiveConsumerWithoutDefinitionProperty() {
System.clearProperty("spring.cloud.stream.function.definition");
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
SingleReactiveConsumerConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false")) {
InputDestination inputDestination = context.getBean(InputDestination.class);
Message<byte[]> inputMessage = MessageBuilder
.withPayload("Hello".getBytes()).build();
inputDestination.send(inputMessage);
assertThat(System.getProperty("consumer")).isEqualTo("Hello");
System.clearProperty("consumer");
}
}
@Test
public void testConsumer() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
@@ -396,6 +415,18 @@ public class ImplicitFunctionBindingTests {
}
}
@EnableAutoConfiguration
public static class SingleReactiveConsumerConfiguration {
@Bean
public Consumer<Flux<String>> consumer() {
return flux -> flux.subscribe(value -> {
System.out.println(value);
System.setProperty("consumer", value);
});
}
}
@EnableAutoConfiguration
public static class ReactiveFunctionConfiguration {