3

我正在使用 RSocket 通道并尝试实现恢复功能，以防与服务器的连接丢失。在尝试了 resume()、resumeSessionDuration()、resumeStreamTimeout() 和 resumeCleanupOnKeepAlive() 的不同组合之后/documentation：https://javadoc.io/static/io.rsocket/rsocket-core/1.0.0-RC3/io/rsocket /RSocketFactory.ServerRSocketFactory.html#resume--/在服务器和移动设备上，仍然没有任何反应。在移动设备上，我每 10 秒向服务器发送一次数据，当连接丢失时，套接字在第 3 次重试时被处理。有谁知道所有这些方法究竟在服务器端和客户端做什么？几乎没有任何文档...

这是服务器上的代码：

RSocketFactory.receive()
                    .resume()
                    .acceptor((setupPayload, reactiveSocket) -> Mono.just(responseHandler))
                    .transport(TcpServerTransport.create(buildSecuredTcpServer(address, port, 
                    sslContext)))
                    .start()
                    .block();

这是移动设备上的代码：

 socket = RSocketFactory.connect()
        .errorConsumer(throwable -> {
            if (throwable instanceof RejectedResumeException){
                Log.d("tagg", "error: " + throwable.getMessage());
            }
        })
        .resume()
        .resumeStreamTimeout(Duration.ofSeconds(120))
        .resumeSessionDuration(Duration.ofSeconds(600))
        .keepAlive(Duration.ofSeconds(120), Duration.ofSeconds(120), 120)
        .resumeStrategy(() -> new PeriodicResumeStrategy(Duration.ofSeconds(1)))
        .metadataMimeType(BuildConfig.RSOCKET_METADATA_MIME_TYPE)
        .dataMimeType(BuildConfig.RSOCKET_DATA_MIME_TYPE)
        .transport(TcpClientTransport.create(tcpClient))
        .start()
        .block();
Objects.requireNonNull(socket)
        .requestChannel(Flux.from(this::onSubscribe))
        .doOnNext(this::receiveData)
        .retryBackoff(Integer.MAX_VALUE, Duration.ofSeconds(1))
        .subscribe();
4

0 回答 0