From 5adc3d66660c3123a5806f168ccdf2298d398aec Mon Sep 17 00:00:00 2001 From: Sebastien Deleuze Date: Wed, 18 Sep 2019 11:39:59 +0200 Subject: [PATCH] Fix RSocketRequester API for requests without payload This commit makes it possible to send requests without requiring to call data(Mono.empty()). It introduces a dedicated MetadataSpec interface and merge ResponseSpec into RequestSpec for more flexibility. Closes gh-23649 --- .../rsocket/DefaultRSocketRequester.java | 65 ++++++++----------- .../messaging/rsocket/RSocketRequester.java | 50 +++++++------- .../rsocket/RSocketRequesterExtensions.kt | 26 ++++---- .../rsocket/DefaultRSocketRequesterTests.java | 5 +- .../RSocketRequesterExtensionsTests.kt | 51 +++++++-------- 5 files changed, 91 insertions(+), 106 deletions(-) diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java index 15e7c5e653..8bdddea5ab 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java @@ -116,6 +116,12 @@ final class DefaultRSocketRequester implements RSocketRequester { private final MetadataEncoder metadataEncoder; + @Nullable + private Mono payloadMono = Mono.empty(); + + @Nullable + private Flux payloadFlux = null; + public DefaultRequestSpec(String route, Object... vars) { this.metadataEncoder = new MetadataEncoder(metadataMimeType(), strategies); @@ -135,24 +141,26 @@ final class DefaultRSocketRequester implements RSocketRequester { } @Override - public RequestSpec metadata(Consumer configurer) { + public RequestSpec metadata(Consumer> configurer) { configurer.accept(this); return this; } @Override - public ResponseSpec data(Object data) { + public RequestSpec data(Object data) { Assert.notNull(data, "'data' must not be null"); - return toResponseSpec(data, ResolvableType.NONE); + createPayload(data, ResolvableType.NONE); + return this; } @Override - public ResponseSpec data(Object producer, Class elementClass) { + public RequestSpec data(Object producer, Class elementClass) { Assert.notNull(producer, "'producer' must not be null"); Assert.notNull(elementClass, "'elementClass' must not be null"); ReactiveAdapter adapter = getAdapter(producer.getClass()); Assert.notNull(adapter, "'producer' type is unknown to ReactiveAdapterRegistry"); - return toResponseSpec(adapter.toPublisher(producer), ResolvableType.forClass(elementClass)); + createPayload(adapter.toPublisher(producer), ResolvableType.forClass(elementClass)); + return this; } @Nullable @@ -161,15 +169,16 @@ final class DefaultRSocketRequester implements RSocketRequester { } @Override - public ResponseSpec data(Object producer, ParameterizedTypeReference elementTypeRef) { + public RequestSpec data(Object producer, ParameterizedTypeReference elementTypeRef) { Assert.notNull(producer, "'producer' must not be null"); Assert.notNull(elementTypeRef, "'elementTypeRef' must not be null"); ReactiveAdapter adapter = getAdapter(producer.getClass()); Assert.notNull(adapter, "'producer' type is unknown to ReactiveAdapterRegistry"); - return toResponseSpec(adapter.toPublisher(producer), ResolvableType.forType(elementTypeRef)); + createPayload(adapter.toPublisher(producer), ResolvableType.forType(elementTypeRef)); + return this; } - private ResponseSpec toResponseSpec(Object input, ResolvableType elementType) { + private void createPayload(Object input, ResolvableType elementType) { ReactiveAdapter adapter = getAdapter(input.getClass()); Publisher publisher; if (input instanceof Publisher) { @@ -179,31 +188,35 @@ final class DefaultRSocketRequester implements RSocketRequester { publisher = adapter.toPublisher(input); } else { - Mono payloadMono = Mono + this.payloadMono = Mono .fromCallable(() -> encodeData(input, ResolvableType.forInstance(input), null)) .map(this::firstPayload) .doOnDiscard(Payload.class, Payload::release) .switchIfEmpty(emptyPayload()); - return new DefaultResponseSpec(payloadMono); + this.payloadFlux = null; + return; } if (isVoid(elementType) || (adapter != null && adapter.isNoValue())) { - Mono payloadMono = Mono.when(publisher).then(emptyPayload()); - return new DefaultResponseSpec(payloadMono); + this.payloadMono = Mono.when(publisher).then(emptyPayload()); + this.payloadFlux = null; + return; } Encoder encoder = elementType != ResolvableType.NONE && !Object.class.equals(elementType.resolve()) ? strategies.encoder(elementType, dataMimeType) : null; if (adapter != null && !adapter.isMultiValue()) { - Mono payloadMono = Mono.from(publisher) + this.payloadMono = Mono.from(publisher) .map(value -> encodeData(value, elementType, encoder)) .map(this::firstPayload) .switchIfEmpty(emptyPayload()); - return new DefaultResponseSpec(payloadMono); + this.payloadFlux = null; + return; } - Flux payloadFlux = Flux.from(publisher) + this.payloadMono = null; + this.payloadFlux = Flux.from(publisher) .map(value -> encodeData(value, elementType, encoder)) .switchOnFirst((signal, inner) -> { DataBuffer data = signal.get(); @@ -217,7 +230,6 @@ final class DefaultRSocketRequester implements RSocketRequester { }) .doOnDiscard(Payload.class, Payload::release) .switchIfEmpty(emptyPayload()); - return new DefaultResponseSpec(payloadFlux); } @SuppressWarnings("unchecked") @@ -245,26 +257,6 @@ final class DefaultRSocketRequester implements RSocketRequester { private Mono emptyPayload() { return Mono.fromCallable(() -> firstPayload(emptyDataBuffer)); } - } - - - private class DefaultResponseSpec implements ResponseSpec { - - @Nullable - private final Mono payloadMono; - - @Nullable - private final Flux payloadFlux; - - DefaultResponseSpec(Mono payloadMono) { - this.payloadMono = payloadMono; - this.payloadFlux = null; - } - - DefaultResponseSpec(Flux payloadFlux) { - this.payloadMono = null; - this.payloadFlux = payloadFlux; - } @Override public Mono send() { @@ -325,5 +317,4 @@ final class DefaultRSocketRequester implements RSocketRequester { return PayloadUtils.retainDataAndReleasePayload(payload, bufferFactory()); } } - } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequester.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequester.java index 5dc770ead9..4998a1f92f 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequester.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequester.java @@ -231,22 +231,10 @@ public interface RSocketRequester { } - /** - * Spec for providing input data for an RSocket request. + * Spec for providing input data for an RSocket request and triggering the exchange. */ - interface RequestSpec { - - /** - * Use this to append additional metadata entries when using composite - * metadata. An {@link IllegalArgumentException} is raised if this - * method is used when not using composite metadata. - * @param metadata an Object to be encoded with a suitable - * {@link org.springframework.core.codec.Encoder Encoder}, or a - * {@link org.springframework.core.io.buffer.DataBuffer DataBuffer} - * @param mimeType the mime type that describes the metadata - */ - RequestSpec metadata(Object metadata, MimeType mimeType); + interface RequestSpec extends MetadataSpec { /** * Append additional metadata entries through a {@code Consumer}. @@ -255,7 +243,7 @@ public interface RSocketRequester { * @param configurer the configurer to apply * @throws IllegalArgumentException if not using composite metadata. */ - RequestSpec metadata(Consumer configurer); + RequestSpec metadata(Consumer> configurer); /** * Provide payload data for the request. This can be one of: @@ -268,7 +256,7 @@ public interface RSocketRequester { * @param data the Object value for the payload data * @return spec to declare the expected response */ - ResponseSpec data(Object data); + RequestSpec data(Object data); /** * Variant of {@link #data(Object)} that also accepts a hint for the @@ -280,7 +268,7 @@ public interface RSocketRequester { * @param elementClass the type of values to be produced * @return spec to declare the expected response */ - ResponseSpec data(Object producer, Class elementClass); + RequestSpec data(Object producer, Class elementClass); /** * Variant of {@link #data(Object, Class)} for when the type hint has @@ -291,14 +279,7 @@ public interface RSocketRequester { * @param elementTypeRef the type of values to be produced * @return spec to declare the expected response */ - ResponseSpec data(Object producer, ParameterizedTypeReference elementTypeRef); - } - - - /** - * Spect to declare the type of request and expected response. - */ - interface ResponseSpec { + RequestSpec data(Object producer, ParameterizedTypeReference elementTypeRef); /** * Perform a {@link RSocket#fireAndForget fireAndForget}. @@ -343,4 +324,23 @@ public interface RSocketRequester { Flux retrieveFlux(ParameterizedTypeReference dataTypeRef); } + /** + * Spec for specifying the metadata. + * + * @param a self reference to the spec type + */ + interface MetadataSpec> { + + /** + * Use this to append additional metadata entries when using composite + * metadata. An {@link IllegalArgumentException} is raised if this + * method is used when not using composite metadata. + * @param metadata an Object to be encoded with a suitable + * {@link org.springframework.core.codec.Encoder Encoder}, or a + * {@link org.springframework.core.io.buffer.DataBuffer DataBuffer} + * @param mimeType the mime type that describes the metadata + */ + S metadata(Object metadata, MimeType mimeType); + } + } diff --git a/spring-messaging/src/main/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensions.kt b/spring-messaging/src/main/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensions.kt index 028a346cce..38b1ed7a52 100644 --- a/spring-messaging/src/main/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensions.kt +++ b/spring-messaging/src/main/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensions.kt @@ -66,7 +66,7 @@ suspend fun RSocketRequester.Builder.connectWebSocketAndAwait(uri: URI): RSocket * @author Sebastien Deleuze * @since 5.2 */ -inline fun RSocketRequester.RequestSpec.dataWithType(producer: Any): RSocketRequester.ResponseSpec = +inline fun RSocketRequester.RequestSpec.dataWithType(producer: Any): RSocketRequester.RequestSpec = data(producer, object : ParameterizedTypeReference() {}) /** @@ -78,7 +78,7 @@ inline fun RSocketRequester.RequestSpec.dataWithType(producer: * @author Sebastien Deleuze * @since 5.2 */ -inline fun RSocketRequester.RequestSpec.dataWithType(publisher: Publisher): RSocketRequester.ResponseSpec = +inline fun RSocketRequester.RequestSpec.dataWithType(publisher: Publisher): RSocketRequester.RequestSpec = data(publisher, object : ParameterizedTypeReference() {}) /** @@ -90,58 +90,58 @@ inline fun RSocketRequester.RequestSpec.dataWithType(publisher * @author Sebastien Deleuze * @since 5.2 */ -inline fun RSocketRequester.RequestSpec.dataWithType(flow: Flow): RSocketRequester.ResponseSpec = +inline fun RSocketRequester.RequestSpec.dataWithType(flow: Flow): RSocketRequester.RequestSpec = data(flow, object : ParameterizedTypeReference() {}) /** - * Coroutines variant of [RSocketRequester.ResponseSpec.send]. + * Coroutines variant of [RSocketRequester.RequestSpec.send]. * * @author Sebastien Deleuze * @since 5.2 */ -suspend fun RSocketRequester.ResponseSpec.sendAndAwait() { +suspend fun RSocketRequester.RequestSpec.sendAndAwait() { send().awaitFirstOrNull() } /** - * Coroutines variant of [RSocketRequester.ResponseSpec.retrieveMono]. + * Coroutines variant of [RSocketRequester.RequestSpec.retrieveMono]. * * @author Sebastien Deleuze * @since 5.2 */ -suspend inline fun RSocketRequester.ResponseSpec.retrieveAndAwait(): T = +suspend inline fun RSocketRequester.RequestSpec.retrieveAndAwait(): T = retrieveMono(object : ParameterizedTypeReference() {}).awaitSingle() /** - * Coroutines variant of [RSocketRequester.ResponseSpec.retrieveFlux]. + * Coroutines variant of [RSocketRequester.RequestSpec.retrieveFlux]. * * @author Sebastien Deleuze * @since 5.2 */ @ExperimentalCoroutinesApi -inline fun RSocketRequester.ResponseSpec.retrieveFlow(): Flow = +inline fun RSocketRequester.RequestSpec.retrieveFlow(): Flow = retrieveFlux(object : ParameterizedTypeReference() {}).asFlow() /** - * Extension for [RSocketRequester.ResponseSpec.retrieveMono] providing a `retrieveMono()` + * Extension for [RSocketRequester.RequestSpec.retrieveMono] providing a `retrieveMono()` * variant leveraging Kotlin reified type parameters. This extension is not subject to type * erasure and retains actual generic type arguments. * * @author Sebastien Deleuze * @since 5.2 */ -inline fun RSocketRequester.ResponseSpec.retrieveMono(): Mono = +inline fun RSocketRequester.RequestSpec.retrieveMono(): Mono = retrieveMono(object : ParameterizedTypeReference() {}) /** - * Extension for [RSocketRequester.ResponseSpec.retrieveFlux] providing a `retrieveFlux()` + * Extension for [RSocketRequester.RequestSpec.retrieveFlux] providing a `retrieveFlux()` * variant leveraging Kotlin reified type parameters. This extension is not subject to type * erasure and retains actual generic type arguments. * * @author Sebastien Deleuze * @since 5.2 */ -inline fun RSocketRequester.ResponseSpec.retrieveFlux(): Flux = +inline fun RSocketRequester.RequestSpec.retrieveFlux(): Flux = retrieveFlux(object : ParameterizedTypeReference() {}) diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterTests.java index 9eb480fdc3..94055e63f5 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterTests.java @@ -38,7 +38,6 @@ import reactor.test.StepVerifier; import org.springframework.core.io.buffer.DefaultDataBufferFactory; import org.springframework.lang.Nullable; import org.springframework.messaging.rsocket.RSocketRequester.RequestSpec; -import org.springframework.messaging.rsocket.RSocketRequester.ResponseSpec; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.assertj.core.api.Assertions.assertThat; @@ -87,7 +86,7 @@ public class DefaultRSocketRequesterTests { testSendMono(spec -> spec.data(Mono.delay(MILLIS_10).then(), Void.class), ""); } - private void testSendMono(Function mapper, String expectedValue) { + private void testSendMono(Function mapper, String expectedValue) { mapper.apply(this.requester.route("toA")).send().block(Duration.ofSeconds(5)); assertThat(this.rsocket.getSavedMethodName()).isEqualTo("fireAndForget"); @@ -111,7 +110,7 @@ public class DefaultRSocketRequesterTests { testSendFlux(spec -> spec.data(stringFlux.cast(Object.class), Object.class), values); } - private void testSendFlux(Function mapper, String... expectedValues) { + private void testSendFlux(Function mapper, String... expectedValues) { this.rsocket.reset(); mapper.apply(this.requester.route("toA")).retrieveFlux(String.class).blockLast(Duration.ofSeconds(5)); diff --git a/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensionsTests.kt b/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensionsTests.kt index 545a48e0a6..341bd5666a 100644 --- a/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensionsTests.kt +++ b/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketRequesterExtensionsTests.kt @@ -74,83 +74,78 @@ class RSocketRequesterExtensionsTests { @Test fun `dataWithType with Publisher`() { val requestSpec = mockk() - val responseSpec = mockk() val data = mockk>() - every { requestSpec.data(any>(), match>(stringTypeRefMatcher)) } returns responseSpec - assertThat(requestSpec.dataWithType(data)).isEqualTo(responseSpec) + every { requestSpec.data(any>(), match>(stringTypeRefMatcher)) } returns requestSpec + assertThat(requestSpec.dataWithType(data)).isEqualTo(requestSpec) } @Test fun `dataWithType with Flow`() { val requestSpec = mockk() - val responseSpec = mockk() val data = mockk>() - every { requestSpec.data(any>(), match>(stringTypeRefMatcher)) } returns responseSpec - assertThat(requestSpec.dataWithType(data)).isEqualTo(responseSpec) + every { requestSpec.data(any>(), match>(stringTypeRefMatcher)) } returns requestSpec + assertThat(requestSpec.dataWithType(data)).isEqualTo(requestSpec) } @Test fun `dataWithType with CompletableFuture`() { val requestSpec = mockk() - val responseSpec = mockk() val data = mockk>() - every { requestSpec.data(any>(), match>(stringTypeRefMatcher)) } returns responseSpec - assertThat(requestSpec.dataWithType(data)).isEqualTo(responseSpec) + every { requestSpec.data(any>(), match>(stringTypeRefMatcher)) } returns requestSpec + assertThat(requestSpec.dataWithType(data)).isEqualTo(requestSpec) } @Test fun dataFlowWithoutType() { val requestSpec = mockk() - val responseSpec = mockk() - every { requestSpec.data(any()) } returns responseSpec - assertThat(requestSpec.data(mockk())).isEqualTo(responseSpec) + every { requestSpec.data(any()) } returns requestSpec + assertThat(requestSpec.data(mockk())).isEqualTo(requestSpec) } @Test fun sendAndAwait() { - val responseSpec = mockk() - every { responseSpec.send() } returns Mono.empty() + val requestSpec = mockk() + every { requestSpec.send() } returns Mono.empty() runBlocking { - responseSpec.sendAndAwait() + requestSpec.sendAndAwait() } } @Test fun retrieveAndAwait() { val response = "foo" - val responseSpec = mockk() - every { responseSpec.retrieveMono(match>(stringTypeRefMatcher)) } returns Mono.just("foo") + val requestSpec = mockk() + every { requestSpec.retrieveMono(match>(stringTypeRefMatcher)) } returns Mono.just("foo") runBlocking { - assertThat(responseSpec.retrieveAndAwait()).isEqualTo(response) + assertThat(requestSpec.retrieveAndAwait()).isEqualTo(response) } } @Test @ExperimentalCoroutinesApi fun retrieveFlow() { - val responseSpec = mockk() - every { responseSpec.retrieveFlux(match>(stringTypeRefMatcher)) } returns Flux.just("foo", "bar") + val requestSpec = mockk() + every { requestSpec.retrieveFlux(match>(stringTypeRefMatcher)) } returns Flux.just("foo", "bar") runBlocking { - assertThat(responseSpec.retrieveFlow().toList()).contains("foo", "bar") + assertThat(requestSpec.retrieveFlow().toList()).contains("foo", "bar") } } @Test fun retrieveMono() { - val responseSpec = mockk() - every { responseSpec.retrieveMono(match>(stringTypeRefMatcher)) } returns Mono.just("foo") + val requestSpec = mockk() + every { requestSpec.retrieveMono(match>(stringTypeRefMatcher)) } returns Mono.just("foo") runBlocking { - assertThat(responseSpec.retrieveMono().block()).isEqualTo("foo") + assertThat(requestSpec.retrieveMono().block()).isEqualTo("foo") } } @Test fun retrieveFlux() { - val responseSpec = mockk() - every { responseSpec.retrieveFlux(match>(stringTypeRefMatcher)) } returns Flux.just("foo", "bar") + val requestSpec = mockk() + every { requestSpec.retrieveFlux(match>(stringTypeRefMatcher)) } returns Flux.just("foo", "bar") runBlocking { - assertThat(responseSpec.retrieveFlux().collectList().block()).contains("foo", "bar") + assertThat(requestSpec.retrieveFlux().collectList().block()).contains("foo", "bar") } } - }