diff --git a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/FunctionRSocketMessageHandler.java b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/FunctionRSocketMessageHandler.java index b67103c53..14aedff59 100644 --- a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/FunctionRSocketMessageHandler.java +++ b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/FunctionRSocketMessageHandler.java @@ -173,7 +173,8 @@ class FunctionRSocketMessageHandler extends RSocketMessageHandler { @Override protected List initArgumentResolvers() { - return Collections.singletonList(new MessageHandlerMethodArgumentResolver(this.jsonMapper)); + List resolvers = super.initArgumentResolvers(); + return Collections.singletonList(new MessageHandlerMethodArgumentResolver(this.jsonMapper, resolvers)); } @SuppressWarnings("unchecked") @@ -225,9 +226,12 @@ class FunctionRSocketMessageHandler extends RSocketMessageHandler { private final JsonMapper jsonMapper; - MessageHandlerMethodArgumentResolver(JsonMapper jsonMapper) { + private final List resolvers; + + MessageHandlerMethodArgumentResolver(JsonMapper jsonMapper, List resolvers) { this.decoder = new ByteArrayDecoder(); this.jsonMapper = jsonMapper; + this.resolvers = resolvers; } @Override @@ -244,25 +248,37 @@ class FunctionRSocketMessageHandler extends RSocketMessageHandler { ? Flux.just((DataBuffer) payload) : Flux.from((Publisher) payload); - Flux decoded = this.decoder.decode(data, ResolvableType.forType(Object.class), null, null) - .map(value -> { - if (JsonMapper.isJsonString(value)) { - // could be array, map or string - Object structure = this.jsonMapper.fromJson(value, Object.class); - if (structure instanceof Map) { - if (((Map) structure).containsKey(FunctionRSocketUtils.PAYLOAD)) { - return MessageBuilder.withPayload(((Map) structure).remove(FunctionRSocketUtils.PAYLOAD)) - .copyHeaders((Map) ((Map) structure).get(FunctionRSocketUtils.HEADERS)) - .build(); - } - else { - return MessageBuilder.withPayload(structure).build(); - } + if (MimeTypeUtils.APPLICATION_JSON_VALUE.equals(message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString())) { + Flux argument = data.map(buffer -> { + byte[] bytePayload = this.decoder.decode(buffer, ResolvableType.forType(byte[].class), null, null); + if (JsonMapper.isJsonString(bytePayload)) { +// // could be array, map or string + Object structure = this.jsonMapper.fromJson(bytePayload, Object.class); + if (structure instanceof Map) { + if (((Map) structure).containsKey(FunctionRSocketUtils.PAYLOAD)) { + return MessageBuilder.withPayload(((Map) structure).remove(FunctionRSocketUtils.PAYLOAD)) + .copyHeaders((Map) ((Map) structure).get(FunctionRSocketUtils.HEADERS)) + .build(); + } + else { + return MessageBuilder.withPayload(structure).build(); } } - return value; - }); - return MessageBuilder.createMessage(decoded, message.getHeaders()); + } + return MessageBuilder.withPayload(bytePayload).copyHeadersIfAbsent(message.getHeaders()).build(); + }); + return MessageBuilder.createMessage(argument, message.getHeaders()); + } + else { + for (HandlerMethodArgumentResolver handlerMethodArgumentResolver : this.resolvers) { + if (handlerMethodArgumentResolver.supportsParameter(parameter)) { + Publisher arg = handlerMethodArgumentResolver.resolveArgument(parameter, message); + return MessageBuilder.withPayload(arg).copyHeadersIfAbsent(message.getHeaders()).build(); + } + + } + return message; + } } } diff --git a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/MessageAwareJsonEncoder.java b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/MessageAwareJsonEncoder.java index 1cc105927..ab1ded943 100644 --- a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/MessageAwareJsonEncoder.java +++ b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/MessageAwareJsonEncoder.java @@ -42,6 +42,9 @@ import org.springframework.util.MimeTypeUtils; * @since 3.1 * */ +/* + * We basically don't need it, but having it allows us not to depend on spring-web + */ class MessageAwareJsonEncoder extends AbstractEncoder { private final JsonMapper mapper; diff --git a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java index b9a137346..1165fd1b9 100644 --- a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java +++ b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java @@ -27,9 +27,11 @@ import reactor.core.publisher.Mono; import org.springframework.cloud.function.context.catalog.FunctionTypeUtils; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.rsocket.annotation.support.RSocketFrameTypeMessageCondition; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; +import org.springframework.util.MimeTypeUtils; /** * A function wrapper which is bound onto an RSocket route. @@ -39,7 +41,7 @@ import org.springframework.util.Assert; * * @since 3.1 */ -class RSocketListenerFunction implements Function>, Publisher> { +class RSocketListenerFunction implements Function> { private final FunctionInvocationWrapper targetFunction; @@ -47,29 +49,38 @@ class RSocketListenerFunction implements Function>, Publish this.targetFunction = targetFunction; } + /* + * We need to maintain the input typeless to ensure that no encoder/decoders will attempt any conversion. + * That said it will always be Message> + */ + @SuppressWarnings("unchecked") @Override - public Publisher apply(Message> input) { + public Publisher apply(Object input) { Assert.isTrue(this.targetFunction != null, "Failed to discover target function. \n" + "To fix it you should either provide 'spring.cloud.function.definition' property " + "or if you are using RSocketRequester provide valid function definition via 'route' " + "operator (e.g., requester.route(\"echo\"))"); - FrameType frameType = RSocketFrameTypeMessageCondition.getFrameType(input); - switch (frameType) { - case REQUEST_FNF: - return handle(input); - case REQUEST_RESPONSE: - case REQUEST_STREAM: - case REQUEST_CHANNEL: - return handleAndReply(input); - default: - throw new UnsupportedOperationException(); - } +// if (input instanceof Message) { + Message> inputMessage = (Message>) input; + FrameType frameType = RSocketFrameTypeMessageCondition.getFrameType(inputMessage); + switch (frameType) { + case REQUEST_FNF: + return handle(inputMessage); + case REQUEST_RESPONSE: + case REQUEST_STREAM: + case REQUEST_CHANNEL: + return handleAndReply(inputMessage); + default: + throw new UnsupportedOperationException(); + } +// } +// throw new UnsupportedOperationException("Expecting input to be of type Message>"); } @SuppressWarnings({ "unchecked", "rawtypes" }) - private Mono handle(Message> messageToProcess) { + private Mono handle(Message> messageToProcess) { if (this.targetFunction.isRoutingFunction()) { - Flux dataFlux = messageToProcess.getPayload() + Flux dataFlux = Flux.from(messageToProcess.getPayload()) .map((payload) -> { return MessageBuilder.createMessage(payload, messageToProcess.getHeaders()); }); @@ -77,7 +88,7 @@ class RSocketListenerFunction implements Function>, Publish } else if (this.targetFunction.isConsumer()) { Flux dataFlux = - messageToProcess.getPayload() + Flux.from(messageToProcess.getPayload()) .map((payload) -> MessageBuilder.createMessage(payload, messageToProcess.getHeaders())); if (FunctionTypeUtils.isPublisher(this.targetFunction.getInputType())) { dataFlux = dataFlux.transform((Function) this.targetFunction); @@ -93,9 +104,9 @@ class RSocketListenerFunction implements Function>, Publish } @SuppressWarnings({ "unchecked", "rawtypes" }) - private Flux handleAndReply(Message> messageToProcess) { + private Flux handleAndReply(Message> messageToProcess) { Flux dataFlux = - messageToProcess.getPayload() + Flux.from(messageToProcess.getPayload()) .map((payload) -> { return payload instanceof Message ? MessageBuilder.fromMessage((Message) payload).copyHeadersIfAbsent(messageToProcess.getHeaders()).build() @@ -111,11 +122,31 @@ class RSocketListenerFunction implements Function>, Publish .copyHeaders((Map) messageMap.get(FunctionRSocketUtils.HEADERS)) .build(); Object result = this.targetFunction.isSupplier() ? this.targetFunction.apply(null) : this.targetFunction.apply(sanitizedMessage); - return result instanceof Publisher + + Publisher resultPublisher = result instanceof Publisher ? (Publisher) result : Mono.just(result); + return Flux.from(resultPublisher).map(v -> extractPayloadIfNecessary(v)); }); } return dataFlux; } + + /* + * This will ensure that unless CT is application/json for which we provide Message aware encoder/decoder + * the payload is extracted since no other available encoders/decoders understand Message. + */ + private Object extractPayloadIfNecessary(Object output) { + if (output instanceof Message) { + Message resultMessage = (Message) output; + Object contentType = resultMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE); + if (contentType != null && contentType.toString().equals(MimeTypeUtils.APPLICATION_JSON_VALUE)) { + return output; + } + else { + return resultMessage.getPayload(); + } + } + return output; + } }