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

- Fix NPE for wildcard/generic type reactive stream listener @Input type

Browse Source 
- Set argumentClass to Object when using wildcard/generic types if the StreamListener method parameter for @Input annotated type uses wildcard/generic type, then set the argumentClass to `Object.class` instead of null.

Resolves #665

Address review comments
This commit is contained in:
Ilayaperumal Gopinathan
2016-10-05 17:53:03 +05:30
committed by Soby Chacko
parent 4e3c43af44
commit 5d2860bc53
2 changed files with 51 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java
View File

@@ -33,6 +33,7 @@ import org.springframework.util.Assert;
* Adapts an {@link org.springframework.cloud.stream.annotation.Input} annotated
* {@link MessageChannel} to a {@link Flux}.
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class MessageChannelToInputFluxParameterAdapter
implements StreamListenerParameterAdapter<Flux<?>, SubscribableChannel> {
@@ -54,7 +55,8 @@ public class MessageChannelToInputFluxParameterAdapter
@Override
public Flux<?> adapt(final SubscribableChannel boundElement, MethodParameter parameter) {
ResolvableType resolvableType = ResolvableType.forMethodParameter(parameter);
Class<?> argumentClass = resolvableType.getGeneric(0).getRawClass();
final Class<?> argumentClass = (resolvableType.getGeneric(0).getRawClass() != null) ? (resolvableType
.getGeneric(0).getRawClass()) : Object.class;
final Object monitor = new Object();
if (Message.class.isAssignableFrom(argumentClass)) {
return Flux.create(emitter -> {
@@ -71,7 +73,8 @@ public class MessageChannelToInputFluxParameterAdapter
return Flux.create(emitter -> {
MessageHandler messageHandler = message -> {
synchronized (monitor) {
if (argumentClass.isAssignableFrom(message.getPayload().getClass())) {
if (argumentClass.isAssignableFrom(message
.getPayload().getClass())) {
emitter.next(message.getPayload());
}
else {

49
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactorTests.java
View File

@@ -38,6 +38,7 @@ import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerReactorTests {
@@ -73,6 +74,22 @@ public class StreamListenerReactorTests {
context.close();
}
@Test
public void testWildCardFluxInputOutputArgsWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestWildCardFluxInputOutputArgsWithMessage.class,
"--server.port=0");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testGenericFluxInputOutputArgsWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestGenericStringFluxInputOutputArgsWithMessage.class,
"--server.port=0");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testInputOutputArgsWithFluxSender() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithFluxSender.class,
@@ -151,9 +168,35 @@ public class StreamListenerReactorTests {
public static class TestInputOutputArgsWithMessage {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<String>> input,
public void receive(@Input(Processor.INPUT) Flux<Message<?>> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder.withPayload(m.getPayload().toUpperCase()).build()));
output.send(input.map(m -> MessageBuilder
.withPayload(m.getPayload().toString().toUpperCase()).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestWildCardFluxInputOutputArgsWithMessage {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<?> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
}
}
public static class TestGenericStringFluxInputOutputArgsWithMessage extends TestGenericFluxInputOutputArgsWithMessage<String> {
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestGenericFluxInputOutputArgsWithMessage<A> {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<A> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder.withPayload((A)m.toString().toUpperCase()).build()));
}
}
@@ -161,7 +204,7 @@ public class StreamListenerReactorTests {
@EnableAutoConfiguration
public static class TestInputOutputArgsWithFluxSender {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<?>> input, @Output(Processor.OUTPUT) FluxSender output) {
public void receive(@Input(Processor.INPUT) Flux<Message<String>> input, @Output(Processor.OUTPUT) FluxSender output) {
output.send(input
.map(m -> m.getPayload().toString().toUpperCase())
.map(o -> MessageBuilder.withPayload(o).build()));