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 11d1a889cb..ef27e95f40 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 @@ -152,17 +152,21 @@ final class DefaultRSocketRequester implements RSocketRequester { } @Override - public > ResponseSpec data(P publisher, Class dataType) { - Assert.notNull(publisher, "'publisher' must not be null"); - Assert.notNull(dataType, "'dataType' must not be null"); - return toResponseSpec(publisher, ResolvableType.forClass(dataType)); + public ResponseSpec data(Object producer, Class elementType) { + Assert.notNull(producer, "'producer' must not be null"); + Assert.notNull(elementType, "'dataType' must not be null"); + ReactiveAdapter adapter = strategies.reactiveAdapterRegistry().getAdapter(producer.getClass()); + Assert.notNull(adapter, "'producer' type is unknown to ReactiveAdapterRegistry"); + return toResponseSpec(adapter.toPublisher(producer), ResolvableType.forClass(elementType)); } @Override - public > ResponseSpec data(P publisher, ParameterizedTypeReference dataTypeRef) { - Assert.notNull(publisher, "'publisher' must not be null"); + public ResponseSpec data(Object producer, ParameterizedTypeReference dataTypeRef) { + Assert.notNull(producer, "'producer' must not be null"); Assert.notNull(dataTypeRef, "'dataTypeRef' must not be null"); - return toResponseSpec(publisher, ResolvableType.forType(dataTypeRef)); + ReactiveAdapter adapter = strategies.reactiveAdapterRegistry().getAdapter(producer.getClass()); + Assert.notNull(adapter, "'producer' type is unknown to ReactiveAdapterRegistry"); + return toResponseSpec(adapter.toPublisher(producer), ResolvableType.forType(dataTypeRef)); } private ResponseSpec toResponseSpec(Object input, ResolvableType dataType) { 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 9689995778..8d7488ada1 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 @@ -200,41 +200,41 @@ public interface RSocketRequester { RequestSpec metadata(Object metadata, MimeType mimeType); /** - * Provide request payload data. The given Object may be a synchronous - * value, or a {@link Publisher} of values, or another async type that's - * registered in the configured {@link ReactiveAdapterRegistry}. - *

For multi-valued Publishers, prefer using - * {@link #data(Publisher, Class)} or - * {@link #data(Publisher, ParameterizedTypeReference)} since that makes - * it possible to find a compatible {@code Encoder} up front vs looking - * it up on every value. + * Provide payload data. The data can be one of the following: + *

    + *
  • Concrete value + *
  • {@link Publisher} of value(s) + *
  • Any other producer of value(s) that can be adapted to a + * {@link Publisher} via {@link ReactiveAdapterRegistry} + *
* @param data the Object to use for payload data * @return spec for declaring the expected response */ ResponseSpec data(Object data); /** - * Provide a {@link Publisher} of value(s) for request payload data. - *

Publisher semantics determined through the configured - * {@link ReactiveAdapterRegistry} influence which of the 4 RSocket - * interactions to use. Publishers with unknown semantics are treated - * as multi-valued. Consider registering a reactive type adapter, or - * passing {@code Mono.from(publisher)}. - *

If the publisher completes empty, possibly {@code Publisher}, - * the request will have an empty data Payload. - * @param publisher source of payload data value(s) - * @param dataType the type of values to be published - * @param the type of element values - * @param

the type of publisher + * Alternative of {@link #data(Object)} that accepts not only a producer + * of value(s) but also a hint for the types of values that will be + * produced. The class hint is used to find a compatible {@code Encoder} + * once, up front, and used for all values. + * @param producer the source of payload data value(s). This must be a + * {@link Publisher} or another producer adaptable to a + * {@code Publisher} via {@link ReactiveAdapterRegistry} + * @param elementType the type of values to be produced * @return spec for declaring the expected response */ - > ResponseSpec data(P publisher, Class dataType); + ResponseSpec data(Object producer, Class elementType); /** - * Variant of {@link #data(Publisher, Class)} for when the dataType has - * to have a generic type. See {@link ParameterizedTypeReference}. + * Alternative of {@link #data(Object, Class)} but with a + * {@link ParameterizedTypeReference} hint which can provide generic + * type information. + * @param producer the source of payload data value(s). This must be a + * {@link Publisher} or another producer adaptable to a + * {@code Publisher} via {@link ReactiveAdapterRegistry} + * @param elementTypeRef the type of values to be produced */ - > ResponseSpec data(P publisher, ParameterizedTypeReference dataTypeRef); + ResponseSpec data(Object producer, ParameterizedTypeReference elementTypeRef); }