如何用RSocket处理服务器发往客户端的消息？

Question

我尝试使用RSocketRequester从服务器向特定的客户端发送消息，但我不知道如何在前端处理它。服务器是Spring Webflux，其控制器如下：

data class Message(val message: String)

@Controller
class RSocketController {

    private val log = LoggerFactory.getLogger(RSocketController::class.java)

    @MessageMapping("say.hello")
    fun sayHello(message: String): Flux<Message> {
        log.info("say hello {}", message)
        return Flux.just(Message("server says hello"))
    }

    @MessageMapping("say.hi")
    fun sayHi(message: String, rSocketRequester: RSocketRequester): Flux<Message> {
        log.info("say hi {}", message)
        rSocketRequester
                .route("say.hello")
                .data(Message("server says hi hello ;)"))
                .send()
                .subscribe()
        return Flux.just(Message("server says hi!!"))
    }

}

在前端，我使用rsocket-js。sayHello方法工作得很好(request-stream)，但是当我调用sayHi方法时，我想从服务器发送两条消息。第一个指向say.hello端点，第二个指向say.hi端点。我有这样的rsocket-js实现：

    sayHello() {
      console.log("say hello");
      this.requestStream("say.hello");
    },
    sayHi() {
      console.log("say hi");
      this.requestStream("say.hi");
    },
    connect() {
      const transport = new RSocketWebSocketClient({
        url: "ws://localhost:8080/rsocket"
      });
      const client = new RSocketClient({        
        serializers: {
          data: JsonSerializer,
          metadata: IdentitySerializer
        },
        setup: {                  
          keepAlive: 60000,          
          lifetime: 180000,
          dataMimeType: "application/json",
          metadataMimeType: "message/x.rsocket.routing.v0"
        },
        transport
      });
      client.connect().subscribe({
        onComplete: socket => {
          this.socket = socket;
          console.log("complete connection");          
        },
        onError: error => {
          console.log("got connection error");
          console.error(error);
        },
        onSubscribe: cancel => {
          console.log("subscribe connection");
          console.log(cancel);          
        }        
      });
    },    
    requestStream(url) {
      if (this.socket) {
        this.socket
          .requestStream({
            data: url + " from client",
            metadata: String.fromCharCode(url.length) + url
          })
          .subscribe({
            onComplete: () => console.log("requestStream done"),
            onError: error => {
              console.log("got error with requestStream");
              console.error(error);
            },
            onNext: value => {
              // console.log("got next value in requestStream..");
              console.log("got data from sever");
              console.log(value.data);
            },
            // Nothing happens until `request(n)` is called
            onSubscribe: sub => {
              console.log("subscribe request Stream!");
              sub.request(2147483647);
              // sub.request(3);
            }
          });
      } else {
        console.log("not connected...");
      }
    }

我可以在谷歌Chrome DevTools ->网络-> rsocket中看到这两条消息。所以客户端接收它们，但我无法在代码中捕获RSocketRequester发送的代码。看起来服务器使用的是fireAndForget方法。如何在客户端处理？

Answer 1

正如@VladMamaev所说，我们可以向客户端提供一个响应器，如本例中的https://github.com/rsocket/rsocket-js/blob/master/packages/rsocket-examples/src/LeaseClientExample.js#L104

对我来说，fireAndForget方法就足够了。

export class EchoResponder {
  constructor(callback) {
    this.callback = callback;
  }
  fireAndForget(payload) {  
    this.callback(payload);
  }  
}

import { EchoResponder } from "~/assets/EchoResponder";
...

  const messageReceiver = payload => {
    //do what you want to do with received message  
    console.log(payload)    
  };
  const responder = new EchoResponder(messageReceiver);


  connect() {
  const transport = new RSocketWebSocketClient({
    url: "ws://localhost:8080/rsocket"
  });
  const client = new RSocketClient({        
    serializers: {
      data: JsonSerializer,
      metadata: IdentitySerializer
    },
    setup: {                  
      keepAlive: 60000,          
      lifetime: 180000,
      dataMimeType: "application/json",
      metadataMimeType: "message/x.rsocket.routing.v0"
    },
    responder: responder,
    transport
  });