Spring Rsocket 双向通道答案

【问题标题】：Spring Rsocket Bi-Directional ChannelSpring Rsocket 双向通道
【发布时间】：2021-04-19 16:20:10
【问题描述】：

是否有可能以像 flatMapSequential 这样的顺序方式从客户端向服务器发送消息？如果我使用，默认行为似乎只是像 flatMap 这样几乎顺序的方式

requester.route("name")
         .data(fluxSendToServer.doOnNext(nr -> log.trace("Send next " + nr.getRequest().getCurrentSequenceValue())))
         .retrieveFlux(ResponseMessageWrapper.class)

并在服务器上记录序列。

【问题讨论】：

任何额外的代码来澄清你的意思？到目前为止，您显示的代码 sn-p 是按顺序工作的
+1 以上评论。这应该映射到通道操作。您提供的数据可以根据需要进行结构化。您可以在调用 data() 之前专门使用 flatMapSequential 来保证它的行为类似于 flatMapSequential。

标签： spring rsocket

【解决方案1】：

感谢您的回答。如果我将客户端的代码更改为

  requester.route("name").data(fluxSendToServer.index().map(indexedTuple->{
if(!indexedTuple.getT2().getRequest().getCurrentSequenceValue().equals(indexedTuple.getT1()+1))
{
throw new IllegalStateException("Orderingproblem in client at index "+indexedTuple.getT1());
 };
return indexedTuple.getT2();
 }))

并在接收服务器方法中使用相同的代码

     @MessageMapping("name")
   Flux<EncryptedResponseMessageWrapper> handleMessageClientToServer(Flux<EncryptedRequestMessageWrapper> 
         fluxReceivedAtServer){
         return fluxReceivedAtServer.index().map(indexedTuple->  {
if(!indexedTuple.getT2().getRequest().getCurrentSequenceValue().equals(indexedTuple.getT1()+1))
{
 throw new IllegalStateException("Orderingproblem at server at index"+indexedTuple.getT1());
};
return indexedTuple.getT2();
})

对于 4000 个元素的通量，我每次都会在随机位置出现 IllegalStateException，例如索引 2175 的服务器上的路由发送消息客户端到服务器排序问题中的示例错误

如果我使用较少的元素，它似乎会重复工作。

【讨论】：

如果您可以在单个测试文件中进行复制，您应该针对 rsocke-java 提出建议。感谢举报。