如何在 Flutter 中路由 RSocket 客户端

Question

这是 java (webflux) 代码。

    public Mono<String> sendMessage(SendMessageRequest requestBody, RSocketRequester rSocketRequester) {
        Long userId = clientManager.getUserIdBySocket(rSocketRequester);

        Chat chat = Chat.builder()
                .chatroomId(requestBody.getChatroomId())
                .createdDate(LocalDateTime.now())
                .lastModifiedDate(LocalDateTime.now())
                .userId(userId)
                .build();

        return chatRepository.save(chat)
                .flatMap(entity ->  chatReadRepository.findByUserIdAndChatroomId(userId, entity.getChatroomId()))
                .flatMapMany(entity -> chatMemberRepository.findAllByChatroomIdAndUserIdNot(entity.getChatroomId(), userId))
                .map(ChatMember::getUserId)
                .map(clientManager::getSocketByUserId)
                .flatMap(socketOptional ->
                    socketOptional.<org.reactivestreams.Publisher<String>>map(socketRequester -> socketRequester.route("chat.receive")
                        .data(requestBody)
                        .send()
                        .thenReturn("Success!"))
                        .orElseGet(() -> Mono.just("fail!"))
                )
                .collectList().thenReturn("Success!");
    }

我在java中使用这段代码

socketRequester -> socketRequester.route("chat.receive")
                        .data(requestBody)
                        .send()
                        .thenReturn("Success!"))
                        .orElseGet(() -> Mono.just("fail!"))

这是我的颤动代码。我使用 rsocket：^1.0.0。

void main() async {
  String jwt = "Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIiwicm9sZXMiOlsiVVNFUiJdLCJpYXQiOjE2NTk1OTg1OTgsImV4cCI6MTY1OTYwODU5OH0.bJHn4IKm6DtnGAQAxyruRb-LJSgKt-72-L9g7JqBtHw";
  var rSocket = await RSocketConnector.create()
      .setupPayload(routeAndDataPayload("socket.acceptor", jwt))
      .connect('tcp://192.168.219.101:8081');

  var payload = await rSocket.requestResponse!(routeAndDataPayload("healthcheck", "data!!!"));

  print(payload.getDataUtf8());

  rSocket.close();

  runApp(const MyApp());
}

我怎样才能收到这条消息？

我想让控制器“chat.receive”路由变得扑朔迷离。

如果是java代码，

@MessageMapping("chat.receive")

但我不知道颤动的@MessageMapping ...

Answer 1

首先，rsocket: ^1.0.0. 似乎无法正常工作，您需要使用

rsocket:
    git:
      url: git@github.com:rsocket/rsocket-dart.git
      ref: master

反而。

然后你可以像下面这样创建一个 Future 作为你的连接持有者：

  final Future<RSocket> _rsocketConnectionStream = RSocketConnector.create()
      .keepAlive(2000, 999999999)
      .connect('ws://192.168.1.188:8080')
      .catchError((error) => print(error));

现在您可以像这样进行 RSokcet 调用：

  Payload routeAndDataPayload(String route, String data) {
    var compositeMetadata =
        CompositeMetadata.fromEntries([RoutingMetadata(route, List.empty())]);
    var metadataBytes = compositeMetadata.toUint8Array();
    var dataBytes = Uint8List.fromList(utf8.encode(data));
    return Payload.from(metadataBytes, dataBytes);
  }

  Stream<String?> fetchSeekersNames(String name, double latitude, double longitude, double searchingRadius) {
    var body = """
    {
      "location": {
          "latitude": $latitude,
          "longitude": $longitude
        },
        "radius": $searchingRadius,
        "dancerPartnerSeekerName": "$name"
    }
  """;
    return _rsocketConnectionStream
        .asStream()
        .asyncExpand((rSocket) => rSocket.requestStream!(
            routeAndDataPayload("/api/XXXX", body)))
        .map((element) => element!.getDataUtf8())
        .doOnError((p0, p1) => print(p1));
  }

没有routeAndDataPayload 方法我做不到。经过几天的研究，我写了它。

据我所知，Dart 中没有 @MessageMapping 等价物。 RSocket 库不支持双向通道。

但是如果你试图让你的 flutter 代码保持 RSocket 基础连接，并在它上面做 request-stream，我没能把它拉出来。

我无法让它在 TCP 上工作。我必须将 webSocket 用于我的 RSocket 底层传输。但是如何在 flutter 代码中保持 websocket/rsocket 连接，....！？

如果您发现了什么，请写在这里。 总而言之，我也是