Generalize RSocketRequester data methods

Replace the Publisher argument in RequestSpec's
data(Publisher, Class<T>) and
data(Publisher, ParameterizedTypeReference<T>) methods with Object thus
allowing any reactive type known to the ReactiveAdapterRegistry to be
passed in directly rather than adapted to Publisher first.

spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java

@@ -152,17 +152,21 @@ final class DefaultRSocketRequester implements RSocketRequester {
 		}
 
 		@Override
-		public <T, P extends Publisher<T>> ResponseSpec data(P publisher, Class<T> 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 <T, P extends Publisher<T>> ResponseSpec data(P publisher, ParameterizedTypeReference<T> 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) {

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}.
-		 * <p>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:
+		 * <ul>
+		 * <li>Concrete value
+		 * <li>{@link Publisher} of value(s)
+		 * <li>Any other producer of value(s) that can be adapted to a
+		 * {@link Publisher} via {@link ReactiveAdapterRegistry}
+		 * </ul>
 		 * @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.
-		 * <p>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)}.
-		 * <p>If the publisher completes empty, possibly {@code Publisher<Void>},
-		 * 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 <T> the type of element values
-		 * @param <P> 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
 		 */
-		<T, P extends Publisher<T>> ResponseSpec data(P publisher, Class<T> 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
 		 */
-		<T, P extends Publisher<T>> ResponseSpec data(P publisher, ParameterizedTypeReference<T> dataTypeRef);
+		ResponseSpec data(Object producer, ParameterizedTypeReference<?> elementTypeRef);
 	}