I wrote a little demo for RSocket message
The problem is that I am unable to access the Rsocket
endpoint,
I get the following exception from the server:
The Client-side: configuration:
@Bean
RSocket rSocket() {
return RSocketFactory.connect()
.mimeType(MimeTypeUtils.APPLICATION_JSON_VALUE, MimeTypeUtils.APPLICATION_JSON_VALUE)
.frameDecoder(PayloadDecoder.ZERO_COPY)
.transport(TcpClientTransport.create(new InetSocketAddress(7500)))
.start()
.block();
}
@Bean
RSocketRequester requester(RSocketStrategies strategies) {
return RSocketRequester.wrap(rSocket(), MimeTypeUtils.APPLICATION_JSON, MimeTypeUtils.APPLICATION_JSON, strategies);
}
controller:
private final RSocketRequester requester;
@GetMapping("/greet/{name}")
public Publisher<GreetingsResponse> greet(@PathVariable String name) {
return requester
.route("hello")
.data(new GreetingsRequest(name))
.retrieveMono(GreetingsResponse.class);
}
The server side(using spring Rsocket): yml:
spring:
rsocket:
server:
port: 7500
transport: tcp
main:
lazy-initialization: true
configuration:
@MessageMapping("hello")
Mono<GreetingsResponse> greet(GreetingsRequest request) {
return Mono.just(new GreetingsResponse("Hello " + request.getName() + " @ " + Instant.now()));
}
I am pretty sure it has something to do with the new wrap
function, RSocketRequester.wrap
as it accepts a new parameter metadataMimeType
, I set it to application/Json,
but it does not seems to work
stackTrace:
org.springframework.messaging.MessageDeliveryException: No handler for destination '' at org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler.handleNoMatch(RSocketMessageHandler.java:312) at org.springframework.messaging.handler.invocation.reactive.AbstractMethodMessageHandler.getHandlerMethod(AbstractMethodMessageHandler.java:445) at org.springframework.messaging.handler.invocation.reactive.AbstractMethodMessageHandler.handleMessage(AbstractMethodMessageHandler.java:417) at org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.lambda$handleAndReply$4(MessagingRSocket.java:173) at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:44) at reactor.core.publisher.Mono.subscribe(Mono.java:3920) at reactor.core.publisher.FluxConcatArray$ConcatArraySubscriber.onComplete(FluxConcatArray.java:207) at reactor.core.publisher.FluxConcatArray.subscribe(FluxConcatArray.java:80) at reactor.core.publisher.MonoFromFluxOperator.subscribe(MonoFromFluxOperator.java:74) at io.rsocket.RSocketResponder.handleRequestResponse(RSocketResponder.java:386) at io.rsocket.RSocketResponder.handleFrame(RSocketResponder.java:298) at reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:238) at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drainRegular(FluxGroupBy.java:554) at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drain(FluxGroupBy.java:630) at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.subscribe(FluxGroupBy.java:696) at reactor.core.publisher.Flux.subscribe(Flux.java:8000) at reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onNext(MonoFlatMapMany.java:184) at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1582) at reactor.core.publisher.MonoProcessor.onNext(MonoProcessor.java:316) at io.rsocket.internal.ClientServerInputMultiplexer.lambda$new$1(ClientServerInputMultiplexer.java:116) at reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) at reactor.core.publisher.FluxGroupBy$GroupByMain.drainLoop(FluxGroupBy.java:380) at reactor.core.publisher.FluxGroupBy$GroupByMain.drain(FluxGroupBy.java:316) at reactor.core.publisher.FluxGroupBy$GroupByMain.onNext(FluxGroupBy.java:201) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:206) at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:322) at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:342) at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:91) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352) at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:328) at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:302) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352) at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1421) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930) at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:697) at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632) at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:511) at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) at java.base/java.lang.Thread.run(Thread.java:834)
Let’s move on to the server-side changes. The first thing to do in the rsocket-server project is to create a collection of RSocketRequester clients by adding a class-level variable to the RSocketController.java class as follows: Next, add a connection handler by adding a new method like this:
The response itself is a stream of messages. Every 5 seconds, the client tells the server its current free memory. Think of this as client telemetry data. For this to work, you must ‘register’ this class with your RSocket connection. You’ll do this in the next section.
Finally, as the client disconnects, it’s RSocket channel’s status becomes “CLOSED” and the client is “DISCONNECTED.” You can stop the rsocket-server process by pressing Ctrl-C in its terminal window.
In fact, the RSocket docs don’t use the terms ‘client’ or ‘server.’ The docs use the terms ‘requester’ and ‘responder’ instead. In RSocket, any component can act as a requester, and any component can act as a responder or even both at the same time.
Which spring version are you using? I had a same issue and I solved it by changing spring-boot-starter-parent 2.2.0.M3.
Here is my source https://github.com/han1448/spring-rsocket-example
Added.
I solved this issue. You need to change mimeType to message/x.rsocket.routing.v0
.
You can get this mimeType from MetadataExtractor.ROUTING
.
@Bean
RSocket rSocket() {
return RSocketFactory.connect()
.mimeType(MetadataExtractor.ROUTING.toString(), MimeTypeUtils.APPLICATION_JSON_VALUE)
.frameDecoder(PayloadDecoder.ZERO_COPY)
.transport(TcpClientTransport.create(new InetSocketAddress(7500)))
.start()
.block();
}
@Bean
RSocketRequester requester(RSocketStrategies strategies) {
return RSocketRequester.wrap(rSocket(), MimeTypeUtils.APPLICATION_JSON, MetadataExtractor.ROUTING, strategies);
}
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With