diff --git a/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/Payload.java b/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/Payload.java index 311776ef2f..db69ece453 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/Payload.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/Payload.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -51,7 +51,11 @@ public @interface Payload { *

This attribute may or may not be supported depending on whether the message being * handled contains a non-primitive Object as its payload or is in serialized form and * requires message conversion. - *

When processing STOMP over WebSocket messages this attribute is not supported. + *

This attribute is not supported for: + *

* @since 4.2 */ @AliasFor("value") 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 e20a3dce5b..1762536eec 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 @@ -102,6 +102,12 @@ final class DefaultRSocketRequester implements RSocketRequester { return this.metadataMimeType; } + @Override + public RSocketStrategies strategies() { + return this.strategies; + } + + @Override public RequestSpec route(String route, Object... vars) { return new DefaultRequestSpec(route, vars); 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 80b18ec170..84475176be 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 @@ -83,6 +83,11 @@ public interface RSocketRequester extends Disposable { */ MimeType metadataMimeType(); + /** + * Return the configured {@link RSocketStrategies}. + */ + RSocketStrategies strategies(); + /** * Begin to specify a new request with the given route to a remote handler. *

The route can be a template with placeholders, e.g. diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/DestinationVariableArgumentResolver.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/DestinationVariableArgumentResolver.java new file mode 100644 index 0000000000..6cc9cabc6e --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/DestinationVariableArgumentResolver.java @@ -0,0 +1,65 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import java.util.Collection; + +import org.springframework.core.MethodParameter; +import org.springframework.lang.Nullable; +import org.springframework.messaging.handler.annotation.DestinationVariable; + +/** + * {@link RSocketServiceArgumentResolver} for a + * {@link DestinationVariable @DestinationVariable} annotated argument. + * + *

The argument is treated as a single route variable, or in case of a + * Collection or an array, as multiple route variables. + * + * @author Rossen Stoyanchev + * @since 6.0 + */ +public class DestinationVariableArgumentResolver implements RSocketServiceArgumentResolver { + + @Override + public boolean resolve( + @Nullable Object argument, MethodParameter parameter, RSocketRequestValues.Builder requestValues) { + + DestinationVariable annot = parameter.getParameterAnnotation(DestinationVariable.class); + if (annot == null) { + return false; + } + + if (argument != null) { + if (argument instanceof Collection) { + ((Collection) argument).forEach(requestValues::addRouteVariable); + return true; + } + else if (argument.getClass().isArray()) { + for (Object variable : (Object[]) argument) { + requestValues.addRouteVariable(variable); + } + return true; + } + else { + requestValues.addRouteVariable(argument); + } + } + + return true; + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/MetadataArgumentResolver.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/MetadataArgumentResolver.java new file mode 100644 index 0000000000..3e89f95bed --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/MetadataArgumentResolver.java @@ -0,0 +1,61 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import org.springframework.core.MethodParameter; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; +import org.springframework.util.MimeType; + +/** + * {@link RSocketServiceArgumentResolver} for metadata entries. + * + *

Supports a sequence of an {@link Object} parameter for the metadata value, + * followed by a {@link MimeType} parameter for the metadata mime type. + * + *

This should be ordered last to give other, more specific resolvers a + * chance to resolve the argument. + * + * @author Rossen Stoyanchev + * @since 6.0 + */ +public class MetadataArgumentResolver implements RSocketServiceArgumentResolver { + + @Override + public boolean resolve( + @Nullable Object argument, MethodParameter parameter, RSocketRequestValues.Builder requestValues) { + + int index = parameter.getParameterIndex(); + Class[] paramTypes = parameter.getExecutable().getParameterTypes(); + + if (parameter.getParameterType().equals(MimeType.class)) { + Assert.notNull(argument, "MimeType parameter is required"); + Assert.state(index > 0, "MimeType parameter should have preceding metadata object parameter"); + requestValues.addMimeType((MimeType) argument); + return true; + } + + if (paramTypes.length > (index + 1) && MimeType.class.equals(paramTypes[index + 1])) { + Assert.notNull(argument, "MimeType parameter is required"); + requestValues.addMetadata(argument); + return true; + } + + return false; + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/PayloadArgumentResolver.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/PayloadArgumentResolver.java new file mode 100644 index 0000000000..8511f675f2 --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/PayloadArgumentResolver.java @@ -0,0 +1,78 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import org.springframework.core.MethodParameter; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.core.ReactiveAdapter; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.lang.Nullable; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.util.Assert; + +/** + * {@link RSocketServiceArgumentResolver} for {@link Payload @Payload} + * annotated arguments. + * + * @author Rossen Stoyanchev + * @since 6.0 + */ +public class PayloadArgumentResolver implements RSocketServiceArgumentResolver { + + private final ReactiveAdapterRegistry reactiveAdapterRegistry; + + private final boolean useDefaultResolution; + + + public PayloadArgumentResolver(ReactiveAdapterRegistry reactiveAdapterRegistry, boolean useDefaultResolution) { + this.useDefaultResolution = useDefaultResolution; + Assert.notNull(reactiveAdapterRegistry, "ReactiveAdapterRegistry is required"); + this.reactiveAdapterRegistry = reactiveAdapterRegistry; + } + + + @Override + public boolean resolve( + @Nullable Object argument, MethodParameter parameter, RSocketRequestValues.Builder requestValues) { + + Payload annot = parameter.getParameterAnnotation(Payload.class); + if (annot == null && !this.useDefaultResolution) { + return false; + } + + if (argument != null) { + ReactiveAdapter reactiveAdapter = this.reactiveAdapterRegistry.getAdapter(parameter.getParameterType()); + if (reactiveAdapter == null) { + requestValues.setPayloadValue(argument); + } + else { + MethodParameter nestedParameter = parameter.nested(); + + String message = "Async type for @Payload should produce value(s)"; + Assert.isTrue(nestedParameter.getNestedParameterType() != Void.class, message); + Assert.isTrue(!reactiveAdapter.isNoValue(), message); + + requestValues.setPayload( + reactiveAdapter.toPublisher(argument), + ParameterizedTypeReference.forType(nestedParameter.getNestedGenericParameterType())); + } + } + + return true; + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketExchange.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketExchange.java new file mode 100644 index 0000000000..115d7bd294 --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketExchange.java @@ -0,0 +1,79 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Annotation to declare a method on an RSocket service interface as an RSocket + * endpoint. The endpoint route is defined statically through the annotation + * attributes, and through the input method argument types. + * + *

Supported at the type level to express common attributes, to be inherited + * by all methods, such as a base route. + * + *

Supported method arguments: + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
Method ArgumentDescriptionResolver
{@link org.springframework.messaging.handler.annotation.DestinationVariable @DestinationVariable}Add a route variable to expand into the route{@link DestinationVariableArgumentResolver}
{@link org.springframework.messaging.handler.annotation.Payload @Payload}Set the input payload(s) for the request{@link PayloadArgumentResolver}
{@link Object} argument followed by {@link org.springframework.util.MimeType} argumentAdd a metadata value{@link MetadataArgumentResolver}
{@link org.springframework.util.MimeType} argument preceded by {@link Object} argumentSpecify the mime type for the preceding metadata value{@link MetadataArgumentResolver}
+ * + * @author Rossen Stoyanchev + * @since 6.0 + */ +@Target({ElementType.TYPE, ElementType.METHOD}) +@Retention(RetentionPolicy.RUNTIME) +@Documented +public @interface RSocketExchange { + + /** + * Destination-based mapping expressed by this annotation. This is either + * {@link org.springframework.util.AntPathMatcher AntPathMatcher} or + * {@link org.springframework.web.util.pattern.PathPattern PathPattern} + * based pattern, depending on which is configured, matched to the route of + * the stream request. + */ + String value() default ""; + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketRequestValues.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketRequestValues.java new file mode 100644 index 0000000000..4ea1e38d9e --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketRequestValues.java @@ -0,0 +1,268 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import org.reactivestreams.Publisher; + +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; +import org.springframework.util.MimeType; +import org.springframework.util.StringUtils; + +/** + * Container for RSocket request values extracted from an + * {@link RSocketExchange @RSocketExchange}-annotated + * method and argument values passed to it. This is then used to define a request + * via {@link org.springframework.messaging.rsocket.RSocketRequester}. + * + * @author Rossen Stoyanchev + * @since 6.0 + */ +public final class RSocketRequestValues { + + @Nullable + private final String route; + + private final Object[] routeVariables; + + private final Map metadata; + + @Nullable + private final Object payloadValue; + + @Nullable + private final Publisher payload; + + @Nullable + private final ParameterizedTypeReference payloadElementType; + + + public RSocketRequestValues( + @Nullable String route, @Nullable List routeVariables, @Nullable MetadataHelper metadataHelper, + @Nullable Object payloadValue, @Nullable Publisher payload, + @Nullable ParameterizedTypeReference payloadElementType) { + + this.route = route; + this.routeVariables = (routeVariables != null ? routeVariables.toArray() : new Object[0]); + this.metadata = (metadataHelper != null ? metadataHelper.toMap() : Collections.emptyMap()); + this.payloadValue = payloadValue; + this.payload = payload; + this.payloadElementType = payloadElementType; + } + + + /** + * Return the route value for + * {@link org.springframework.messaging.rsocket.RSocketRequester#route(String, Object...) route}. + */ + @Nullable + public String getRoute() { + return this.route; + } + + /** + * Return the route variables for + * {@link org.springframework.messaging.rsocket.RSocketRequester#route(String, Object...) route}. + */ + public Object[] getRouteVariables() { + return this.routeVariables; + } + + /** + * Return the metadata entries for + * {@link org.springframework.messaging.rsocket.RSocketRequester.RequestSpec#metadata(Object, MimeType)}. + */ + public Map getMetadata() { + return this.metadata; + } + + /** + * Return the request payload as a value to be serialized, if set. + *

This is mutually exclusive with {@link #getPayload()}. + * Only one of the two or neither is set. + */ + @Nullable + public Object getPayloadValue() { + return this.payloadValue; + } + + /** + * Return the request payload as a Publisher. + *

This is mutually exclusive with {@link #getPayloadValue()}. + * Only one of the two or neither is set. + */ + @Nullable + public Publisher getPayload() { + return this.payload; + } + + /** + * Return the element type for a {@linkplain #getPayload() Publisher payload}. + */ + @Nullable + public ParameterizedTypeReference getPayloadElementType() { + return this.payloadElementType; + } + + + public static Builder builder(@Nullable String route) { + return new Builder(route); + } + + + /** + * Builder for {@link RSocketRequestValues}. + */ + public final static class Builder { + + @Nullable + private String route; + + @Nullable + private List routeVariables; + + @Nullable + private MetadataHelper metadataHelper; + + @Nullable + private Object payloadValue; + + @Nullable + private Publisher payload; + + @Nullable + private ParameterizedTypeReference payloadElementType; + + Builder(@Nullable String route) { + this.route = (StringUtils.hasText(route) ? route : null); + } + + /** + * Set the route for the request. + */ + public Builder setRoute(String route) { + this.route = route; + this.routeVariables = null; + return this; + } + + /** + * Add a route variable. + */ + public Builder addRouteVariable(Object variable) { + this.routeVariables = (this.routeVariables != null ? this.routeVariables : new ArrayList<>()); + this.routeVariables.add(variable); + return this; + } + + /** + * Add a metadata entry. + * This must be followed by a corresponding call to {@link #addMimeType(MimeType)}. + */ + public Builder addMetadata(Object metadata) { + this.metadataHelper = (this.metadataHelper != null ? this.metadataHelper : new MetadataHelper()); + this.metadataHelper.addMetadata(metadata); + return this; + } + + /** + * Set the mime type for a metadata entry. + * This must be preceded by a call to {@link #addMetadata(Object)}. + */ + public Builder addMimeType(MimeType mimeType) { + this.metadataHelper = (this.metadataHelper != null ? this.metadataHelper : new MetadataHelper()); + this.metadataHelper.addMimeType(mimeType); + return this; + } + + /** + * Set the request payload as a concrete value to be serialized. + *

This is mutually exclusive with, and resets any previously set + * {@linkplain #setPayload(Publisher, ParameterizedTypeReference) payload Publisher}. + */ + public Builder setPayloadValue(Object payloadValue) { + this.payloadValue = payloadValue; + this.payload = null; + this.payloadElementType = null; + return this; + } + + /** + * Set the request payload value to be serialized. + */ + public > Builder setPayload(P payload, ParameterizedTypeReference elementTye) { + this.payload = payload; + this.payloadElementType = elementTye; + this.payloadValue = null; + return this; + } + + /** + * Build the {@link RSocketRequestValues} instance. + */ + public RSocketRequestValues build() { + return new RSocketRequestValues( + this.route, this.routeVariables, this.metadataHelper, + this.payloadValue, this.payload, this.payloadElementType); + } + + } + + + /** + * Class that helps to collect a map of metadata entries as a series of calls + * to provide each metadata and mime type pair. + */ + private static class MetadataHelper { + + private final List metadata = new ArrayList<>(); + + private final List mimeTypes = new ArrayList<>(); + + public void addMetadata(Object metadata) { + Assert.isTrue(this.metadata.size() == this.mimeTypes.size(), "Invalid state: " + this); + this.metadata.add(metadata); + } + + public void addMimeType(MimeType mimeType) { + Assert.isTrue(this.metadata.size() == (this.mimeTypes.size() + 1), "Invalid state: " + this); + this.mimeTypes.add(mimeType); + } + + public Map toMap() { + Map map = new LinkedHashMap<>(this.metadata.size()); + for (int i = 0; i < this.metadata.size(); i++) { + map.put(this.metadata.get(i), this.mimeTypes.get(i)); + } + return map; + } + + @Override + public String toString() { + return "metadata=" + this.metadata + ", mimeTypes=" + this.mimeTypes; + } + + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketServiceArgumentResolver.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketServiceArgumentResolver.java new file mode 100644 index 0000000000..a15b97e86b --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketServiceArgumentResolver.java @@ -0,0 +1,40 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import org.springframework.core.MethodParameter; +import org.springframework.lang.Nullable; + +/** + * Resolve an argument from an {@link RSocketExchange @RSocketExchange}-annotated + * method to one or more RSocket request values. + * + * @author Rossen Stoyanchev + * @since 6.0 + */ +public interface RSocketServiceArgumentResolver { + + /** + * Resolve the argument value. + * @param argument the argument value + * @param parameter the method parameter for the argument + * @param requestValues builder to add RSocket request values to + * @return {@code true} if the argument was resolved, {@code false} otherwise + */ + boolean resolve(@Nullable Object argument, MethodParameter parameter, RSocketRequestValues.Builder requestValues); + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketServiceMethod.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketServiceMethod.java new file mode 100644 index 0000000000..57b4f8daba --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketServiceMethod.java @@ -0,0 +1,242 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import java.lang.reflect.Method; +import java.time.Duration; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; + +import org.reactivestreams.Publisher; +import reactor.core.publisher.Mono; + +import org.springframework.core.DefaultParameterNameDiscoverer; +import org.springframework.core.MethodParameter; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.core.ReactiveAdapter; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.core.annotation.AnnotatedElementUtils; +import org.springframework.core.annotation.SynthesizingMethodParameter; +import org.springframework.lang.Nullable; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.util.Assert; +import org.springframework.util.MimeType; +import org.springframework.util.StringUtils; +import org.springframework.util.StringValueResolver; + +/** + * Implements the invocation of an {@link RSocketExchange @RSocketExchange}-annotated, + * {@link RSocketServiceProxyFactory#createClient(Class) RSocket service proxy} method + * by delegating to an {@link RSocketRequester} to perform actual requests. + * + * @author Rossen Stoyanchev + * @since 6.0 + */ +final class RSocketServiceMethod { + + private final Method method; + + private final MethodParameter[] parameters; + + private final List argumentResolvers; + + @Nullable + private final String route; + + private final Function responseFunction; + + + RSocketServiceMethod( + Method method, Class containingClass, List argumentResolvers, + RSocketRequester rsocketRequester, @Nullable StringValueResolver embeddedValueResolver, + ReactiveAdapterRegistry reactiveRegistry, Duration blockTimeout) { + + this.method = method; + this.parameters = initMethodParameters(method); + this.argumentResolvers = argumentResolvers; + this.route = initRoute(method, containingClass, rsocketRequester.strategies(), embeddedValueResolver); + this.responseFunction = initResponseFunction( + rsocketRequester, method, reactiveRegistry, blockTimeout); + } + + private static MethodParameter[] initMethodParameters(Method method) { + int count = method.getParameterCount(); + if (count == 0) { + return new MethodParameter[0]; + } + DefaultParameterNameDiscoverer nameDiscoverer = new DefaultParameterNameDiscoverer(); + MethodParameter[] parameters = new MethodParameter[count]; + for (int i = 0; i < count; i++) { + parameters[i] = new SynthesizingMethodParameter(method, i); + parameters[i].initParameterNameDiscovery(nameDiscoverer); + } + return parameters; + } + + @Nullable + private static String initRoute( + Method method, Class containingClass, RSocketStrategies strategies, + @Nullable StringValueResolver embeddedValueResolver) { + + RSocketExchange annot1 = AnnotatedElementUtils.findMergedAnnotation(containingClass, RSocketExchange.class); + RSocketExchange annot2 = AnnotatedElementUtils.findMergedAnnotation(method, RSocketExchange.class); + + Assert.notNull(annot2, "Expected RSocketExchange annotation"); + + String route1 = (annot1 != null ? annot1.value() : null); + String route2 = annot2.value(); + + if (embeddedValueResolver != null) { + route1 = (route1 != null ? embeddedValueResolver.resolveStringValue(route1) : null); + route2 = embeddedValueResolver.resolveStringValue(route2); + } + + boolean hasRoute1 = StringUtils.hasText(route1); + boolean hasRoute2 = StringUtils.hasText(route2); + + if (hasRoute1 && hasRoute2) { + return strategies.routeMatcher().combine(route1, route2); + } + + if (!hasRoute1 && !hasRoute2) { + return null; + } + + return (hasRoute2 ? route2 : route1); + } + + private static Function initResponseFunction( + RSocketRequester requester, Method method, + ReactiveAdapterRegistry reactiveRegistry, Duration blockTimeout) { + + MethodParameter returnParam = new MethodParameter(method, -1); + Class returnType = returnParam.getParameterType(); + ReactiveAdapter reactiveAdapter = reactiveRegistry.getAdapter(returnType); + + MethodParameter actualParam = (reactiveAdapter != null ? returnParam.nested() : returnParam.nestedIfOptional()); + Class actualType = actualParam.getNestedParameterType(); + + Function> responseFunction; + if (actualType.equals(void.class) || actualType.equals(Void.class) || + (reactiveAdapter != null && reactiveAdapter.isNoValue())) { + + responseFunction = values -> { + RSocketRequester.RetrieveSpec retrieveSpec = initRequest(requester, values); + return (values.getPayload() == null && values.getPayloadValue() == null ? + ((RSocketRequester.RequestSpec) retrieveSpec).sendMetadata() : retrieveSpec.send()); + }; + } + else if (reactiveAdapter == null) { + responseFunction = values -> initRequest(requester, values).retrieveMono(actualType); + } + else { + ParameterizedTypeReference payloadType = + ParameterizedTypeReference.forType(actualParam.getNestedGenericParameterType()); + + responseFunction = values -> ( + reactiveAdapter.isMultiValue() ? + initRequest(requester, values).retrieveFlux(payloadType) : + initRequest(requester, values).retrieveMono(payloadType)); + } + + boolean blockForOptional = returnType.equals(Optional.class); + + return responseFunction.andThen(responsePublisher -> { + if (reactiveAdapter != null) { + return reactiveAdapter.fromPublisher(responsePublisher); + } + return (blockForOptional ? + ((Mono) responsePublisher).blockOptional(blockTimeout) : + ((Mono) responsePublisher).block(blockTimeout)); + }); + } + + @SuppressWarnings("ReactiveStreamsUnusedPublisher") + private static RSocketRequester.RetrieveSpec initRequest( + RSocketRequester requester, RSocketRequestValues requestValues) { + + RSocketRequester.RequestSpec spec; + String route = requestValues.getRoute(); + Map metadata = requestValues.getMetadata(); + + if (StringUtils.hasText(route)) { + spec = requester.route(route, requestValues.getRouteVariables()); + for (Map.Entry entry : metadata.entrySet()) { + spec.metadata(entry.getKey(), entry.getValue()); + } + } + else { + Iterator> iterator = metadata.entrySet().iterator(); + Assert.isTrue(iterator.hasNext(), "Neither route nor metadata provided"); + + Map.Entry entry = iterator.next(); + spec = requester.metadata(entry.getKey(), entry.getValue()); + + while (iterator.hasNext()) { + spec.metadata(entry.getKey(), entry.getValue()); + } + } + + if (requestValues.getPayloadValue() != null) { + spec.data(requestValues.getPayloadValue()); + } + else if (requestValues.getPayload() != null) { + Assert.notNull(requestValues.getPayloadElementType(), "Publisher body element type is required"); + spec.data(requestValues.getPayload(), requestValues.getPayloadElementType()); + } + + return spec; + } + + + public Method getMethod() { + return this.method; + } + + @Nullable + public Object invoke(Object[] arguments) { + RSocketRequestValues.Builder requestValues = RSocketRequestValues.builder(this.route); + applyArguments(requestValues, arguments); + return this.responseFunction.apply(requestValues.build()); + } + + private void applyArguments(RSocketRequestValues.Builder requestValues, Object[] arguments) { + Assert.isTrue(arguments.length == this.parameters.length, "Method argument mismatch"); + for (int i = 0; i < arguments.length; i++) { + Object value = arguments[i]; + boolean resolved = false; + for (RSocketServiceArgumentResolver resolver : this.argumentResolvers) { + if (resolver.resolve(value, this.parameters[i], requestValues)) { + resolved = true; + break; + } + } + Assert.state(resolved, formatArgumentError(this.parameters[i], "No suitable resolver")); + } + } + + @SuppressWarnings("SameParameterValue") + private static String formatArgumentError(MethodParameter param, String message) { + return "Could not resolve parameter [" + param.getParameterIndex() + "] in " + + param.getExecutable().toGenericString() + (StringUtils.hasText(message) ? ": " + message : ""); + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketServiceProxyFactory.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketServiceProxyFactory.java new file mode 100644 index 0000000000..500e07045f --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/RSocketServiceProxyFactory.java @@ -0,0 +1,217 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import java.util.stream.Collectors; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; + +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.aop.framework.ReflectiveMethodInvocation; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.EmbeddedValueResolverAware; +import org.springframework.core.MethodIntrospector; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.core.annotation.AnnotatedElementUtils; +import org.springframework.lang.Nullable; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.util.Assert; +import org.springframework.util.StringValueResolver; + +/** + * Factory for creating a client proxy given an RSocket service interface with + * {@link RSocketExchange @RSocketExchange} methods. + * + *

This class is intended to be declared as a bean in Spring configuration. + * + * @author Rossen Stoyanchev + * @since 6.0 + */ +public final class RSocketServiceProxyFactory implements InitializingBean, EmbeddedValueResolverAware { + + private final RSocketRequester rsocketRequester; + + @Nullable + private List customArgumentResolvers; + + @Nullable + private List argumentResolvers; + + @Nullable + private StringValueResolver embeddedValueResolver; + + private ReactiveAdapterRegistry reactiveAdapterRegistry = ReactiveAdapterRegistry.getSharedInstance(); + + private Duration blockTimeout = Duration.ofSeconds(5); + + + /** + * Create an instance with the underlying RSocketRequester to perform requests with. + * @param rsocketRequester the requester to use + */ + public RSocketServiceProxyFactory(RSocketRequester rsocketRequester) { + Assert.notNull(rsocketRequester, "RSocketRequester is required"); + this.rsocketRequester = rsocketRequester; + } + + + /** + * Register a custom argument resolver, invoked ahead of default resolvers. + * @param resolver the resolver to add + */ + public void addCustomArgumentResolver(RSocketServiceArgumentResolver resolver) { + if (this.customArgumentResolvers == null) { + this.customArgumentResolvers = new ArrayList<>(); + } + this.customArgumentResolvers.add(resolver); + } + + /** + * Set the custom argument resolvers to use, ahead of default resolvers. + * @param resolvers the resolvers to use + */ + public void setCustomArgumentResolvers(List resolvers) { + this.customArgumentResolvers = new ArrayList<>(resolvers); + } + + /** + * Set the StringValueResolver to use for resolving placeholders and + * expressions in {@link RSocketExchange#value()}. + * @param resolver the resolver to use + */ + @Override + public void setEmbeddedValueResolver(StringValueResolver resolver) { + this.embeddedValueResolver = resolver; + } + + /** + * Set the {@link ReactiveAdapterRegistry} to use to support different + * asynchronous types for RSocket service method return values. + *

By default this is {@link ReactiveAdapterRegistry#getSharedInstance()}. + */ + public void setReactiveAdapterRegistry(ReactiveAdapterRegistry registry) { + this.reactiveAdapterRegistry = registry; + } + + /** + * Configure how long to wait for a response for an RSocket service method + * with a synchronous (blocking) method signature. + *

By default this is 5 seconds. + * @param blockTimeout the timeout value + */ + public void setBlockTimeout(Duration blockTimeout) { + this.blockTimeout = blockTimeout; + } + + + @Override + public void afterPropertiesSet() throws Exception { + this.argumentResolvers = initArgumentResolvers(); + } + + private List initArgumentResolvers() { + List resolvers = new ArrayList<>(); + + // Custom + if (this.customArgumentResolvers != null) { + resolvers.addAll(this.customArgumentResolvers); + } + + // Annotation-based + resolvers.add(new PayloadArgumentResolver(this.reactiveAdapterRegistry, false)); + resolvers.add(new DestinationVariableArgumentResolver()); + + // Type-based + resolvers.add(new MetadataArgumentResolver()); + + // Fallback + resolvers.add(new PayloadArgumentResolver(this.reactiveAdapterRegistry, true)); + + return resolvers; + } + + + /** + * Return a proxy that implements the given RSocket service interface to + * perform RSocket requests and retrieve responses through the configured + * {@link RSocketRequester}. + * @param serviceType the RSocket service to create a proxy for + * @param the RSocket service type + * @return the created proxy + */ + public S createClient(Class serviceType) { + + List serviceMethods = + MethodIntrospector.selectMethods(serviceType, this::isExchangeMethod).stream() + .map(method -> createRSocketServiceMethod(serviceType, method)) + .toList(); + + return ProxyFactory.getProxy(serviceType, new ServiceMethodInterceptor(serviceMethods)); + } + + private boolean isExchangeMethod(Method method) { + return AnnotatedElementUtils.hasAnnotation(method, RSocketExchange.class); + } + + private RSocketServiceMethod createRSocketServiceMethod(Class serviceType, Method method) { + Assert.notNull(this.argumentResolvers, + "No argument resolvers: afterPropertiesSet was not called"); + + return new RSocketServiceMethod( + method, serviceType, this.argumentResolvers, this.rsocketRequester, + this.embeddedValueResolver, this.reactiveAdapterRegistry, this.blockTimeout); + } + + + /** + * {@link MethodInterceptor} that invokes an {@link RSocketServiceMethod}. + */ + private static final class ServiceMethodInterceptor implements MethodInterceptor { + + private final Map serviceMethods; + + private ServiceMethodInterceptor(List methods) { + this.serviceMethods = methods.stream() + .collect(Collectors.toMap(RSocketServiceMethod::getMethod, Function.identity())); + } + + @Override + public Object invoke(MethodInvocation invocation) throws Throwable { + Method method = invocation.getMethod(); + RSocketServiceMethod serviceMethod = this.serviceMethods.get(method); + if (serviceMethod != null) { + return serviceMethod.invoke(invocation.getArguments()); + } + if (method.isDefault()) { + if (invocation instanceof ReflectiveMethodInvocation reflectiveMethodInvocation) { + Object proxy = reflectiveMethodInvocation.getProxy(); + return InvocationHandler.invokeDefault(proxy, method, invocation.getArguments()); + } + } + throw new IllegalStateException("Unexpected method invocation: " + method); + } + } + +} diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/package-info.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/package-info.java new file mode 100644 index 0000000000..65c6ba8c22 --- /dev/null +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/service/package-info.java @@ -0,0 +1,11 @@ +/** + * Annotations to declare an RSocket service contract with request methods along + * with a proxy factory backed by an + * {@link org.springframework.messaging.rsocket.RSocketRequester}. + */ +@NonNullApi +@NonNullFields +package org.springframework.messaging.rsocket.service; + +import org.springframework.lang.NonNullApi; +import org.springframework.lang.NonNullFields; 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 9032be5328..7761e1d4a0 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 @@ -28,17 +28,14 @@ import io.reactivex.rxjava3.core.Completable; import io.reactivex.rxjava3.core.Observable; import io.reactivex.rxjava3.core.Single; import io.rsocket.Payload; -import io.rsocket.RSocket; import io.rsocket.metadata.WellKnownMimeType; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; 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.RetrieveSpec; import org.springframework.util.MimeType; @@ -246,73 +243,4 @@ public class DefaultRSocketRequesterTests { return PayloadUtils.createPayload(DefaultDataBufferFactory.sharedInstance.wrap(bytes)); } - - private static class TestRSocket implements RSocket { - - private Mono payloadMonoToReturn = Mono.empty(); - private Flux payloadFluxToReturn = Flux.empty(); - - @Nullable private volatile String savedMethodName; - @Nullable private volatile Payload savedPayload; - @Nullable private volatile Flux savedPayloadFlux; - - void setPayloadMonoToReturn(Mono payloadMonoToReturn) { - this.payloadMonoToReturn = payloadMonoToReturn; - } - - void setPayloadFluxToReturn(Flux payloadFluxToReturn) { - this.payloadFluxToReturn = payloadFluxToReturn; - } - - @Nullable - String getSavedMethodName() { - return this.savedMethodName; - } - - @Nullable - Payload getSavedPayload() { - return this.savedPayload; - } - - @Nullable - Flux getSavedPayloadFlux() { - return this.savedPayloadFlux; - } - - public void reset() { - this.savedMethodName = null; - this.savedPayload = null; - this.savedPayloadFlux = null; - } - - - @Override - public Mono fireAndForget(Payload payload) { - this.savedMethodName = "fireAndForget"; - this.savedPayload = payload; - return Mono.empty(); - } - - @Override - public Mono requestResponse(Payload payload) { - this.savedMethodName = "requestResponse"; - this.savedPayload = payload; - return this.payloadMonoToReturn; - } - - @Override - public Flux requestStream(Payload payload) { - this.savedMethodName = "requestStream"; - this.savedPayload = payload; - return this.payloadFluxToReturn; - } - - @Override - public Flux requestChannel(Publisher publisher) { - this.savedMethodName = "requestChannel"; - this.savedPayloadFlux = Flux.from(publisher); - return this.payloadFluxToReturn; - } - } - } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/TestRSocket.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/TestRSocket.java new file mode 100644 index 0000000000..15580a3026 --- /dev/null +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/TestRSocket.java @@ -0,0 +1,101 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket; + +import io.rsocket.Payload; +import io.rsocket.RSocket; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.lang.Nullable; + +/** + * {@link RSocket} that saves the name of the invoked method and the input payload(s). + */ +public class TestRSocket implements RSocket { + + private Mono payloadMonoToReturn = Mono.empty(); + + private Flux payloadFluxToReturn = Flux.empty(); + + @Nullable private volatile String savedMethodName; + + @Nullable private volatile Payload savedPayload; + + @Nullable private volatile Flux savedPayloadFlux; + + + public void setPayloadMonoToReturn(Mono payloadMonoToReturn) { + this.payloadMonoToReturn = payloadMonoToReturn; + } + + public void setPayloadFluxToReturn(Flux payloadFluxToReturn) { + this.payloadFluxToReturn = payloadFluxToReturn; + } + + @Nullable + public String getSavedMethodName() { + return this.savedMethodName; + } + + @Nullable + public Payload getSavedPayload() { + return this.savedPayload; + } + + @Nullable + public Flux getSavedPayloadFlux() { + return this.savedPayloadFlux; + } + + public void reset() { + this.savedMethodName = null; + this.savedPayload = null; + this.savedPayloadFlux = null; + } + + + @Override + public Mono fireAndForget(Payload payload) { + this.savedMethodName = "fireAndForget"; + this.savedPayload = payload; + return Mono.empty(); + } + + @Override + public Mono requestResponse(Payload payload) { + this.savedMethodName = "requestResponse"; + this.savedPayload = payload; + return this.payloadMonoToReturn; + } + + @Override + public Flux requestStream(Payload payload) { + this.savedMethodName = "requestStream"; + this.savedPayload = payload; + return this.payloadFluxToReturn; + } + + @Override + public Flux requestChannel(Publisher publisher) { + this.savedMethodName = "requestChannel"; + this.savedPayloadFlux = Flux.from(publisher); + return this.payloadFluxToReturn; + } + +} diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/DestinationVariableArgumentResolverTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/DestinationVariableArgumentResolverTests.java new file mode 100644 index 0000000000..3dd093a628 --- /dev/null +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/DestinationVariableArgumentResolverTests.java @@ -0,0 +1,99 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import java.util.Arrays; +import java.util.List; + +import org.junit.jupiter.api.Test; + +import org.springframework.core.MethodParameter; +import org.springframework.messaging.handler.annotation.DestinationVariable; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit tests for {@link DestinationVariableArgumentResolver}. + * @author Rossen Stoyanchev + */ +public class DestinationVariableArgumentResolverTests extends RSocketServiceArgumentResolverTestSupport { + + @Override + protected RSocketServiceArgumentResolver initResolver() { + return new DestinationVariableArgumentResolver(); + } + + @Test + void variable() { + String value = "foo"; + boolean resolved = execute(value, initMethodParameter(Service.class, "execute", 0)); + + assertThat(resolved).isTrue(); + assertThat(getRequestValues().getRouteVariables()).containsExactly(value); + } + + @Test + void variableList() { + List values = Arrays.asList("foo", "bar", "baz"); + boolean resolved = execute(values, initMethodParameter(Service.class, "execute", 0)); + + assertThat(resolved).isTrue(); + assertThat(getRequestValues().getRouteVariables()).containsExactlyElementsOf(values); + } + + @Test + void variableArray() { + String[] values = new String[] {"foo", "bar", "baz"}; + boolean resolved = execute(values, initMethodParameter(Service.class, "execute", 0)); + + assertThat(resolved).isTrue(); + assertThat(getRequestValues().getRouteVariables()).containsExactlyElementsOf(Arrays.asList(values)); + } + + @Test + void notRequestBody() { + MethodParameter parameter = initMethodParameter(Service.class, "executeNotAnnotated", 0); + boolean resolved = execute("value", parameter); + + assertThat(resolved).isFalse(); + } + + @Test + void ignoreNull() { + boolean resolved = execute(null, initMethodParameter(Service.class, "execute", 0)); + + assertThat(resolved).isTrue(); + assertThat(getRequestValues().getPayloadValue()).isNull(); + assertThat(getRequestValues().getPayload()).isNull(); + assertThat(getRequestValues().getPayloadElementType()).isNull(); + } + + + @SuppressWarnings("unused") + private interface Service { + + void execute(@DestinationVariable String variable); + + void executeList(@DestinationVariable List variables); + + void executeArray(@DestinationVariable String[] variables); + + void executeNotAnnotated(String variable); + + } + +} diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/MetadataArgumentResolverTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/MetadataArgumentResolverTests.java new file mode 100644 index 0000000000..894cba5a3d --- /dev/null +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/MetadataArgumentResolverTests.java @@ -0,0 +1,71 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import java.util.LinkedHashMap; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import org.springframework.core.MethodParameter; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit tests for {@link MetadataArgumentResolver}. + * @author Rossen Stoyanchev + */ +public class MetadataArgumentResolverTests extends RSocketServiceArgumentResolverTestSupport { + + @Override + protected RSocketServiceArgumentResolver initResolver() { + return new MetadataArgumentResolver(); + } + + @Test + void metadata() { + MethodParameter param1 = initMethodParameter(Service.class, "execute", 0); + MethodParameter param2 = initMethodParameter(Service.class, "execute", 1); + MethodParameter param3 = initMethodParameter(Service.class, "execute", 2); + MethodParameter param4 = initMethodParameter(Service.class, "execute", 3); + + assertThat(execute("foo", param1)).isTrue(); + assertThat(execute(MimeTypeUtils.APPLICATION_JSON, param2)).isTrue(); + + assertThat(execute("bar", param3)).isTrue(); + assertThat(execute(MimeTypeUtils.APPLICATION_XML, param4)).isTrue(); + + Map expected = new LinkedHashMap<>(); + expected.put("foo", MimeTypeUtils.APPLICATION_JSON); + expected.put("bar", MimeTypeUtils.APPLICATION_XML); + + assertThat(getRequestValues().getMetadata()).containsExactlyEntriesOf(expected); + } + + + @SuppressWarnings("unused") + private interface Service { + + void execute(String metadata1, MimeType mimeType1, String metadata2, MimeType mimeType2); + + void executeNotAnnotated(String foo, String bar); + + } + +} diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/PayloadArgumentResolverTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/PayloadArgumentResolverTests.java new file mode 100644 index 0000000000..d585d64bac --- /dev/null +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/PayloadArgumentResolverTests.java @@ -0,0 +1,129 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import io.reactivex.rxjava3.core.Completable; +import io.reactivex.rxjava3.core.Single; +import org.junit.jupiter.api.Test; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Mono; + +import org.springframework.core.MethodParameter; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.messaging.handler.annotation.Payload; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; + +/** + * Unit tests for {@link PayloadArgumentResolver}. + * @author Rossen Stoyanchev + */ +public class PayloadArgumentResolverTests extends RSocketServiceArgumentResolverTestSupport { + + @Override + protected RSocketServiceArgumentResolver initResolver() { + return new PayloadArgumentResolver(ReactiveAdapterRegistry.getSharedInstance(), false); + } + + @Test + void stringPayload() { + String payload = "payloadValue"; + boolean resolved = execute(payload, initMethodParameter(Service.class, "execute", 0)); + + assertThat(resolved).isTrue(); + assertThat(getRequestValues().getPayloadValue()).isEqualTo(payload); + assertThat(getRequestValues().getPayload()).isNull(); + } + + @Test + void monoPayload() { + Mono payloadMono = Mono.just("payloadValue"); + boolean resolved = execute(payloadMono, initMethodParameter(Service.class, "executeMono", 0)); + + assertThat(resolved).isTrue(); + assertThat(getRequestValues().getPayloadValue()).isNull(); + assertThat(getRequestValues().getPayload()).isSameAs(payloadMono); + assertThat(getRequestValues().getPayloadElementType()).isEqualTo(new ParameterizedTypeReference() {}); + } + + @Test + @SuppressWarnings("unchecked") + void singlePayload() { + boolean resolved = execute(Single.just("bodyValue"), initMethodParameter(Service.class, "executeSingle", 0)); + + assertThat(resolved).isTrue(); + assertThat(getRequestValues().getPayloadValue()).isNull(); + assertThat(getRequestValues().getPayloadElementType()).isEqualTo(new ParameterizedTypeReference() {}); + + Publisher payload = getRequestValues().getPayload(); + assertThat(payload).isNotNull(); + assertThat(((Mono) payload).block()).isEqualTo("bodyValue"); + } + + @Test + void monoVoid() { + assertThatIllegalArgumentException() + .isThrownBy(() -> execute(Mono.empty(), initMethodParameter(Service.class, "executeMonoVoid", 0))) + .withMessage("Async type for @Payload should produce value(s)"); + } + + @Test + void completable() { + assertThatIllegalArgumentException() + .isThrownBy(() -> execute(Completable.complete(), initMethodParameter(Service.class, "executeCompletable", 0))) + .withMessage("Async type for @Payload should produce value(s)"); + } + + @Test + void notRequestBody() { + MethodParameter parameter = initMethodParameter(Service.class, "executeNotAnnotated", 0); + boolean resolved = execute("value", parameter); + + assertThat(resolved).isFalse(); + } + + @Test + void ignoreNull() { + boolean resolved = execute(null, initMethodParameter(Service.class, "execute", 0)); + + assertThat(resolved).isTrue(); + assertThat(getRequestValues().getPayloadValue()).isNull(); + assertThat(getRequestValues().getPayload()).isNull(); + assertThat(getRequestValues().getPayloadElementType()).isNull(); + } + + + @SuppressWarnings("unused") + private interface Service { + + void execute(@Payload String body); + + void executeMono(@Payload Mono body); + + void executeSingle(@Payload Single body); + + void executeMonoVoid(@Payload Mono body); + + void executeCompletable(@Payload Completable body); + + void executeNotAnnotated(String body); + + } + +} diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketRequestValuesTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketRequestValuesTests.java new file mode 100644 index 0000000000..defffb15e5 --- /dev/null +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketRequestValuesTests.java @@ -0,0 +1,95 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + + +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Mono; + +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.util.MimeTypeUtils; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; + +/** + * Unit tests for {@link RSocketRequestValues}. + * @author Rossen Stoyanchev + */ +public class RSocketRequestValuesTests { + + @Test + void route() { + String myRoute = "myRoute"; + RSocketRequestValues values = RSocketRequestValues.builder(myRoute).build(); + assertThat(values.getRoute()).isEqualTo(myRoute); + } + + @Test + void routeOverride() { + RSocketRequestValues values = RSocketRequestValues.builder("route1").setRoute("route2").build(); + + assertThat(values.getRoute()).isEqualTo("route2"); + } + + @Test + void payloadValue() { + String payload = "myValue"; + RSocketRequestValues values = RSocketRequestValues.builder(null).setPayloadValue(payload).build(); + + assertThat(values.getPayloadValue()).isEqualTo(payload); + assertThat(values.getPayload()).isNull(); + } + + @Test + void payloadPublisher() { + Mono payloadMono = Mono.just( "myValue"); + RSocketRequestValues values = RSocketRequestValues.builder(null) + .setPayload(payloadMono, new ParameterizedTypeReference<>() { }) + .build(); + + assertThat(values.getPayloadValue()).isNull(); + assertThat(values.getPayload()).isSameAs(payloadMono); + } + + @Test + void metadata() { + RSocketRequestValues values = RSocketRequestValues.builder(null) + .addMetadata("myMetadata1").addMimeType(MimeTypeUtils.TEXT_PLAIN) + .addMetadata("myMetadata2").addMimeType(MimeTypeUtils.TEXT_HTML) + .build(); + + assertThat(values.getMetadata()) + .hasSize(2) + .containsEntry("myMetadata1", MimeTypeUtils.TEXT_PLAIN) + .containsEntry("myMetadata2", MimeTypeUtils.TEXT_HTML); + } + + @Test + void metadataInvalidEntry() { + // MimeType without metadata + assertThatIllegalArgumentException() + .isThrownBy(() -> RSocketRequestValues.builder(null).addMimeType(MimeTypeUtils.TEXT_PLAIN)); + + // Metadata without MimeType + assertThatIllegalArgumentException() + .isThrownBy(() -> RSocketRequestValues.builder(null) + .addMetadata("metadata1") + .addMetadata("metadata2")); + } + +} diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketServiceArgumentResolverTestSupport.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketServiceArgumentResolverTestSupport.java new file mode 100644 index 0000000000..7b534dcca8 --- /dev/null +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketServiceArgumentResolverTestSupport.java @@ -0,0 +1,60 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import java.lang.reflect.Method; + +import org.springframework.core.MethodParameter; +import org.springframework.lang.Nullable; +import org.springframework.util.ClassUtils; + +/** + * Base class for {@link RSocketServiceArgumentResolver} test fixtures. + * @author Rossen Stoyanchev + */ +public abstract class RSocketServiceArgumentResolverTestSupport { + + @Nullable + private RSocketServiceArgumentResolver resolver; + + private final RSocketRequestValues.Builder requestValuesBuilder = RSocketRequestValues.builder(null); + + @Nullable + private RSocketRequestValues requestValues; + + + protected RSocketServiceArgumentResolverTestSupport() { + this.resolver = initResolver(); + } + + protected abstract RSocketServiceArgumentResolver initResolver(); + + protected static MethodParameter initMethodParameter(Class serviceClass, String methodName, int index) { + Method method = ClassUtils.getMethod(serviceClass, methodName, (Class[]) null); + return new MethodParameter(method, index); + } + + protected boolean execute(Object payload, MethodParameter parameter) { + return this.resolver.resolve(payload, parameter, this.requestValuesBuilder); + } + + protected RSocketRequestValues getRequestValues() { + this.requestValues = (this.requestValues != null ? this.requestValues : this.requestValuesBuilder.build()); + return this.requestValues; + } + +} diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketServiceIntegrationTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketServiceIntegrationTests.java new file mode 100644 index 0000000000..e48df0ca62 --- /dev/null +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketServiceIntegrationTests.java @@ -0,0 +1,162 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import java.time.Duration; + +import io.rsocket.SocketAcceptor; +import io.rsocket.core.RSocketServer; +import io.rsocket.metadata.WellKnownMimeType; +import io.rsocket.transport.netty.server.CloseableChannel; +import io.rsocket.transport.netty.server.TcpServerTransport; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.handler.annotation.MessageMapping; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler; +import org.springframework.stereotype.Controller; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; + +/** + * Integration tests with RSocket Service client. + * + * @author Rossen Stoyanchev + */ +public class RSocketServiceIntegrationTests { + + private static CloseableChannel server; + + private static RSocketRequester requester; + + private static Service serviceProxy; + + + @BeforeAll + @SuppressWarnings("ConstantConditions") + public static void setupOnce() throws Exception { + + MimeType metadataMimeType = MimeTypeUtils.parseMimeType( + WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString()); + + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(ServerConfig.class); + RSocketMessageHandler messageHandler = context.getBean(RSocketMessageHandler.class); + SocketAcceptor responder = messageHandler.responder(); + + server = RSocketServer.create(responder) + .bind(TcpServerTransport.create("localhost", 7000)) + .block(); + + requester = RSocketRequester.builder() + .metadataMimeType(metadataMimeType) + .rsocketStrategies(context.getBean(RSocketStrategies.class)) + .tcp("localhost", 7000); + + RSocketServiceProxyFactory proxyFactory = new RSocketServiceProxyFactory(requester); + proxyFactory.afterPropertiesSet(); + + serviceProxy = proxyFactory.createClient(Service.class); + } + + @AfterAll + public static void tearDownOnce() { + requester.rsocketClient().dispose(); + server.dispose(); + } + + + @Test + public void echoAsync() { + Flux result = Flux.range(1, 3).concatMap(i -> serviceProxy.echoAsync("Hello " + i)); + + StepVerifier.create(result) + .expectNext("Hello 1 async").expectNext("Hello 2 async").expectNext("Hello 3 async") + .expectComplete() + .verify(Duration.ofSeconds(5)); + } + + @Test + public void echoStream() { + Flux result = serviceProxy.echoStream("Hello"); + + StepVerifier.create(result) + .expectNext("Hello 0").expectNextCount(6).expectNext("Hello 7") + .thenCancel() + .verify(Duration.ofSeconds(5)); + } + + + @Controller + interface Service { + + @RSocketExchange("echo-async") + Mono echoAsync(String payload); + + @RSocketExchange("echo-stream") + Flux echoStream(String payload); + + } + + + @Controller + static class ServerController { + + @MessageMapping("echo-async") + Mono echoAsync(String payload) { + return Mono.delay(Duration.ofMillis(10)).map(aLong -> payload + " async"); + } + + @MessageMapping("echo-stream") + Flux echoStream(String payload) { + return Flux.interval(Duration.ofMillis(10)).map(aLong -> payload + " " + aLong); + } + + } + + + @Configuration + static class ServerConfig { + + @Bean + public ServerController controller() { + return new ServerController(); + } + + @Bean + public RSocketMessageHandler messageHandler(RSocketStrategies rsocketStrategies) { + RSocketMessageHandler handler = new RSocketMessageHandler(); + handler.setRSocketStrategies(rsocketStrategies); + return handler; + } + + @Bean + public RSocketStrategies rsocketStrategies() { + return RSocketStrategies.create(); + } + + } + +} diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketServiceMethodTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketServiceMethodTests.java new file mode 100644 index 0000000000..905e64a73d --- /dev/null +++ b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/service/RSocketServiceMethodTests.java @@ -0,0 +1,150 @@ +/* + * Copyright 2002-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.messaging.rsocket.service; + +import java.time.Duration; +import java.util.List; + +import io.rsocket.util.DefaultPayload; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.messaging.rsocket.TestRSocket; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.util.MimeTypeUtils.TEXT_PLAIN; + +/** + * Tests for {@link RSocketServiceMethod} that create an interface client with + * an {@link RSocketRequester} delegating to a {@link TestRSocket}. + * + * @author Rossen Stoyanchev + */ +public class RSocketServiceMethodTests { + + private TestRSocket rsocket; + + private RSocketServiceProxyFactory proxyFactory; + + + @BeforeEach + public void setUp() throws Exception { + this.rsocket = new TestRSocket(); + RSocketRequester requester = RSocketRequester.wrap(this.rsocket, TEXT_PLAIN, TEXT_PLAIN, RSocketStrategies.create()); + this.proxyFactory = new RSocketServiceProxyFactory(requester); + this.proxyFactory.afterPropertiesSet(); + } + + + @Test + void fireAndForget() { + ReactorService service = this.proxyFactory.createClient(ReactorService.class); + + String payload = "p1"; + service.fireAndForget(Mono.just(payload)).block(Duration.ofSeconds(5)); + + assertThat(this.rsocket.getSavedMethodName()).isEqualTo("fireAndForget"); + assertThat(this.rsocket.getSavedPayload().getMetadataUtf8()).isEqualTo("ff"); + assertThat(this.rsocket.getSavedPayload().getDataUtf8()).isEqualTo(payload); + } + + @Test + void requestResponse() { + ReactorService service = this.proxyFactory.createClient(ReactorService.class); + + String payload1 = "p1"; + String payload2 = "p2"; + this.rsocket.setPayloadMonoToReturn( + Mono.just(DefaultPayload.create(payload2))); + + String response = service.requestResponse(Mono.just(payload1)).block(Duration.ofSeconds(5)); + + assertThat(response).isEqualTo(payload2); + assertThat(this.rsocket.getSavedMethodName()).isEqualTo("requestResponse"); + assertThat(this.rsocket.getSavedPayload().getMetadataUtf8()).isEqualTo("rr"); + assertThat(this.rsocket.getSavedPayload().getDataUtf8()).isEqualTo(payload1); + } + + @Test + void requestStream() { + ReactorService service = this.proxyFactory.createClient(ReactorService.class); + + String payload1 = "p1"; + String payload2 = "p2"; + String payload3 = "p3"; + this.rsocket.setPayloadFluxToReturn( + Flux.just(DefaultPayload.create(payload2), DefaultPayload.create(payload3))); + + List response = service.requestStream(Mono.just(payload1)) + .collectList() + .block(Duration.ofSeconds(5)); + + assertThat(response).containsExactly(payload2, payload3); + assertThat(this.rsocket.getSavedMethodName()).isEqualTo("requestStream"); + assertThat(this.rsocket.getSavedPayload().getMetadataUtf8()).isEqualTo("rs"); + assertThat(this.rsocket.getSavedPayload().getDataUtf8()).isEqualTo(payload1); + } + + @Test + void requestChannel() { + ReactorService service = this.proxyFactory.createClient(ReactorService.class); + + String payload1 = "p1"; + String payload2 = "p2"; + String payload3 = "p3"; + String payload4 = "p4"; + this.rsocket.setPayloadFluxToReturn( + Flux.just(DefaultPayload.create(payload3), DefaultPayload.create(payload4))); + + List response = service.requestChannel(Flux.just(payload1, payload2)) + .collectList() + .block(Duration.ofSeconds(5)); + + assertThat(response).containsExactly(payload3, payload4); + assertThat(this.rsocket.getSavedMethodName()).isEqualTo("requestChannel"); + + List savedPayloads = this.rsocket.getSavedPayloadFlux() + .map(io.rsocket.Payload::getDataUtf8) + .collectList() + .block(Duration.ofSeconds(5)); + + assertThat(savedPayloads).containsExactly("p1", "p2"); + } + + + private interface ReactorService { + + @RSocketExchange("ff") + Mono fireAndForget(@Payload Mono input); + + @RSocketExchange("rr") + Mono requestResponse(@Payload Mono input); + + @RSocketExchange("rs") + Flux requestStream(@Payload Mono input); + + @RSocketExchange("rc") + Flux requestChannel(@Payload Flux input); + + } + +} diff --git a/spring-web/src/main/java/org/springframework/web/service/invoker/RequestBodyArgumentResolver.java b/spring-web/src/main/java/org/springframework/web/service/invoker/RequestBodyArgumentResolver.java index 2303e5eac4..51004cd931 100644 --- a/spring-web/src/main/java/org/springframework/web/service/invoker/RequestBodyArgumentResolver.java +++ b/spring-web/src/main/java/org/springframework/web/service/invoker/RequestBodyArgumentResolver.java @@ -55,31 +55,23 @@ public class RequestBodyArgumentResolver implements HttpServiceArgumentResolver if (argument != null) { ReactiveAdapter reactiveAdapter = this.reactiveAdapterRegistry.getAdapter(parameter.getParameterType()); - if (reactiveAdapter != null) { - setBody(argument, parameter, reactiveAdapter, requestValues); + if (reactiveAdapter == null) { + requestValues.setBodyValue(argument); } else { - requestValues.setBodyValue(argument); + MethodParameter nestedParameter = parameter.nested(); + + String message = "Async type for @RequestBody should produce value(s)"; + Assert.isTrue(!reactiveAdapter.isNoValue(), message); + Assert.isTrue(nestedParameter.getNestedParameterType() != Void.class, message); + + requestValues.setBody( + reactiveAdapter.toPublisher(argument), + ParameterizedTypeReference.forType(nestedParameter.getNestedGenericParameterType())); } } return true; } - private void setBody( - Object argument, MethodParameter parameter, ReactiveAdapter reactiveAdapter, - HttpRequestValues.Builder requestValues) { - - String message = "Async type for @RequestBody should produce value(s)"; - Assert.isTrue(!reactiveAdapter.isNoValue(), message); - - parameter = parameter.nested(); - Class elementClass = parameter.getNestedParameterType(); - Assert.isTrue(elementClass != Void.class, message); - ParameterizedTypeReference typeRef = ParameterizedTypeReference.forType(parameter.getNestedGenericParameterType()); - Publisher publisher = reactiveAdapter.toPublisher(argument); - - requestValues.setBody(publisher, typeRef); - } - }