Add RSocketServiceMethod support for Kotlin suspending functions
#35473
Conversation
spring-projects-issues
added
the
status: waiting-for-triage
dmitrysulman
changed the title
RSocketServiceMethod support for suspending functionsRSocketServiceMethod support for Kotlin suspending functions
|
@dmitrysulman hello, dear Dmitry. Can you please also add test cases for streaming? At least requestStream that return Flow. Because I also was doing poc of this feature (just for interest) and Flow return type most probably will be covered by ReactiveAdapterRegistry, but still it is good for it to also be covered by test. And also meanwhile I am not sure if the requestChannel will correctly handle Flow as argument, so also good to be covered with test. Will be very grateful if you have time to add it and thank you for your contribution. |
dmitrysulman
force-pushed
the
rsocket-interface-coroutine
branch
from
fee43d6 to
df0dde9
Compare
See spring-projects#34868 Signed-off-by: Dmitry Sulman <[email protected]>
dmitrysulman
force-pushed
the
rsocket-interface-coroutine
branch
from
350082c to
255ef56
Compare
|
@doxlik done, Request Stream and Request Channel suspending functions (using The |
sdeleuze
added
in: messaging
| MethodParameter returnParam = new MethodParameter(method, -1); | ||
| Class<?> returnType = returnParam.getParameterType(); | ||
| boolean isUnwrapped = KotlinDetector.isSuspendingFunction(method) && | ||
| !COROUTINES_FLOW_CLASS_NAME.equals(returnParam.getParameterType().getName()); |
There was a problem hiding this comment.
returnParam.getParameterType().getName() can be replaced with returnType.getName().
@sdeleuze I can update the branch or it can be fixed in the polishing commit.
| suspend fun requestResponse(input: String): String | ||
|
|
||
| @RSocketExchange("rs") | ||
| suspend fun requestStream(input: String): Flow<String> |
There was a problem hiding this comment.
@dmitrysulman By the way, here it is tricky part whether to let suspend functions for "Flow returning" methods (requestStream / requestChannel) because the Flow itself does not require to be in suspend function, all asynchronous work happens in flow {} / Flow.collect {} callback.
And despite I personally prefer to keep such functions also suspend, the big open source thinks other way
- Spring HTTP Interfaces does not allow (at least in 6.x.x.) suspend for "Flow returning" methods.
- GRPC Kotlin also generates non-suspend functions for "Flow returning" methods.
@sdeleuze please correct me if I'm am wrong
There was a problem hiding this comment.
Spring HTTP Interfaces does not allow (at least in 6.x.x.) suspend for "Flow returning" methods.
This can be easily fixed after (if) the current PR is merged (the fix of the CoroutinesUtils.asFlow method is required).
This PR updates
RSocketServiceMethodto allow using the@RSocketExchangeannotation with Kotlin suspending functions when used as RSocket requester.Fixes #34868