From bc1787310e17e2b5b73a1acb8084080ab140b3ff Mon Sep 17 00:00:00 2001 From: Ryan Dunckel Date: Tue, 4 Jun 2019 10:39:24 -0500 Subject: [PATCH] passes a conversion hint to the message converter to support nested generics. Fixes #1723 Resolves #1725 --- spring-cloud-stream-reactive/pom.xml | 5 + ...ageChannelToInputFluxParameterAdapter.java | 8 +- ...cationJsonMessageMarshallingConverter.java | 26 +++ ...annelToInputFluxParameterAdapterTests.java | 154 +++++++++++++++++- 4 files changed, 188 insertions(+), 5 deletions(-) create mode 100644 spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/converter/TestApplicationJsonMessageMarshallingConverter.java diff --git a/spring-cloud-stream-reactive/pom.xml b/spring-cloud-stream-reactive/pom.xml index 46548c9c1..3dda9afcc 100644 --- a/spring-cloud-stream-reactive/pom.xml +++ b/spring-cloud-stream-reactive/pom.xml @@ -35,6 +35,11 @@ spring-cloud-stream-test-support-internal test + + io.projectreactor + reactor-test + test + org.springframework.boot spring-boot-autoconfigure-processor diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java index 9080c3408..27a43a0ad 100644 --- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java +++ b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java @@ -20,6 +20,7 @@ import reactor.core.publisher.Flux; import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter; import org.springframework.core.MethodParameter; +import org.springframework.core.ParameterizedTypeReference; import org.springframework.core.ResolvableType; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -66,6 +67,9 @@ public class MessageChannelToInputFluxParameterAdapter final Object monitor = new Object(); + Object conversionHint = ParameterizedTypeReference + .forType(fluxTypeParameter.getType()); + if (Message.class.isAssignableFrom(fluxTypeParameterClass)) { final ResolvableType payloadTypeParameter = fluxTypeParameter.getGeneric(0); @@ -85,7 +89,7 @@ public class MessageChannelToInputFluxParameterAdapter else { emitter.next(MessageBuilder.createMessage( this.messageConverter.fromMessage(message, - payloadTypeParameterClass), + payloadTypeParameterClass, conversionHint), message.getHeaders())); } } @@ -104,7 +108,7 @@ public class MessageChannelToInputFluxParameterAdapter } else { emitter.next(this.messageConverter.fromMessage(message, - fluxTypeParameterClass)); + fluxTypeParameterClass, conversionHint)); } } }; diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/converter/TestApplicationJsonMessageMarshallingConverter.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/converter/TestApplicationJsonMessageMarshallingConverter.java new file mode 100644 index 000000000..c90747cb4 --- /dev/null +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/converter/TestApplicationJsonMessageMarshallingConverter.java @@ -0,0 +1,26 @@ +package org.springframework.cloud.stream.converter; + +import org.springframework.cloud.stream.reactive.MessageChannelToInputFluxParameterAdapterTests; + +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.MapperFeature; +import com.fasterxml.jackson.databind.ObjectMapper; + +/** + * This class exists because the {@link ApplicationJsonMessageMarshallingConverter} is package-private scoped, + * so it can't be used directly in, e.g. the {@link MessageChannelToInputFluxParameterAdapterTests} class. + * In that case, we are just extending it and making it public (but only for tests). + */ +public class TestApplicationJsonMessageMarshallingConverter + extends ApplicationJsonMessageMarshallingConverter { + + public TestApplicationJsonMessageMarshallingConverter() { + this(new ObjectMapper().configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false) + .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)); + } + + public TestApplicationJsonMessageMarshallingConverter(ObjectMapper objectMapper) { + super(objectMapper); + } + +} diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapterTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapterTests.java index b871df9b3..3433572b1 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapterTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapterTests.java @@ -16,17 +16,20 @@ package org.springframework.cloud.stream.reactive; +import static org.assertj.core.api.Assertions.assertThat; + import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Objects; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.junit.Test; -import reactor.core.publisher.Flux; - +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.converter.TestApplicationJsonMessageMarshallingConverter; import org.springframework.core.MethodParameter; import org.springframework.integration.channel.DirectChannel; import org.springframework.messaging.Message; @@ -35,7 +38,8 @@ import org.springframework.messaging.converter.MappingJackson2MessageConverter; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.ReflectionUtils; -import static org.assertj.core.api.Assertions.assertThat; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; /** * @author Marius Bogoevici @@ -77,8 +81,152 @@ public class MessageChannelToInputFluxParameterAdapterTests { } + @Test + public void testAdapterConvertsUsingConversionHint() { + CompositeMessageConverter messageConverter = new CompositeMessageConverter( + Collections + .singleton(new TestApplicationJsonMessageMarshallingConverter())); + + MessageChannelToInputFluxParameterAdapter adapter = new MessageChannelToInputFluxParameterAdapter( + messageConverter); + + Method processMethod = ReflectionUtils.findMethod( + MessageChannelToInputFluxParameterAdapterTests.class, "processNestedGenericFlux", + Flux.class); + + DirectChannel adaptedChannel = new DirectChannel(); + + @SuppressWarnings("unchecked") + Flux>> adapterFlux = (Flux>>) adapter + .adapt(adaptedChannel, new MethodParameter(processMethod, 0)); + + SecondLevelWrapper expected2 = new SecondLevelWrapper<>(); + expected2.setName("name"); + expected2.setData("data"); + + FirstLevelWrapper> expected1 = new FirstLevelWrapper<>(); + expected1.setId(1); + expected1.setData(expected2); + + StepVerifier.create(adapterFlux).then(() -> { + adaptedChannel.send(MessageBuilder.withPayload( + "{ \"id\": 1, \"data\": { \"name\": \"name\", \"data\": \"data\" } }") + .build()); + }).expectNext(expected1).thenCancel().verify(); + } + public void process(Flux> message) { // do nothing - we just reference this method from the test } + public void processNestedGenericFlux(Flux>> items) { + // do nothing - we just reference this method from the test + } + + static class FirstLevelWrapper { + + private long id; + + private T data; + + long getId() { + return id; + } + + void setId(long id) { + this.id = id; + } + + T getData() { + return data; + } + + public void setData(T data) { + this.data = data; + } + + @Override + public int hashCode() { + return Objects.hash(data, id); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + + if (obj == null) { + return false; + } + + if (!(obj instanceof FirstLevelWrapper)) { + return false; + } + + @SuppressWarnings("unchecked") + FirstLevelWrapper other = (FirstLevelWrapper) obj; + return Objects.equals(data, other.data) && id == other.id; + } + + @Override + public String toString() { + return "FirstLevelWrapper [id=" + id + ", data=" + data + "]"; + } + + } + + static class SecondLevelWrapper { + + private String name; + + private T data; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public T getData() { + return data; + } + + public void setData(T data) { + this.data = data; + } + + @Override + public int hashCode() { + return Objects.hash(data, name); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + + if (obj == null) { + return false; + } + + if (!(obj instanceof SecondLevelWrapper)) { + return false; + } + + @SuppressWarnings("unchecked") + SecondLevelWrapper other = (SecondLevelWrapper) obj; + return Objects.equals(data, other.data) && Objects.equals(name, other.name); + } + + @Override + public String toString() { + return "SecondLevelWrapper [name=" + name + ", data=" + data + "]"; + } + + } + }