有没有办法从Netty中的channel.write（）返回自定义promise？

Question

我目前正在努力实现一种保护隐私的数据挖掘算法。对于不同各方之间的通信部分，我使用的是Netty 4.0。双方之间的通信流如下所示：

         -- multiplicationMsg --> ... -- multiplicationMsg -->
   P_{1}                                                       P_{N}
         <-- multiplicationMsg -- ... <-- multiplicationMsg --

其中P_{1}是启动和控制整个计算的主方。安全多方乘法的逻辑位于 Netty 通道处理程序中。还有另一种用于安全添加的协议。

目前，我使用类似的解决方案，如Netty核心团队的Norman Maurer所示，以了解子协议计算是否已完成。但这感觉有点像是在对抗框架。

是否有办法从< code>channel.write(msg)获得自定义promise，并在< code>ChannelPipeline中创建和履行？在我上面的例子中，当< code>multiplicationMsg返回到< code>P_{1}时，它应该被满足。

编辑1

这是我通常从ChannelPipeline外部写入消息的方式：

ChannelFuture f = channel.write(msg);
future.addListener(new ChannelFutureListener() {
    public void operationComplete(ChannelFuture future) {
         //do something with the future
    }
});

如果数据可以写入套接字或发生故障，则上述示例中的ChannelFuturef将得到满足。但我需要一种方法来取回自定义的Future除了ChannelFuture，类似于：

ChannelFuture f = channel.write(msg);
future.addListener(new ChannelFutureListener() {
    public void operationComplete(ChannelFuture future) {
         // I need something like the following
         if(future.isSuccess()) {
             Future myFuture = future.getMyFuture();
         }
    }
});

Answer 1

有很多方法可以做到这一点，下面是一个在netty之上html" target="_blank">构建的示例：

在管道外部，使用包含< code>ChannelFuture(来自连接初始化)的类(比如< code>IoClient)内的公共方法发送消息。该方法类似于以下内容:

public MyCustomFuture send(String msg) {
  MyCustomFuture responseFuture = new MyCustomFuture();

  channelFuture.channel().pipeline().get(MyAppClientHandler.class).setResponseFuture(responseFuture);
  channelFuture.channel().writeAndFlush(msg);   

  return responseFuture;
}

MyCustomFuture 是我们创建的自定义类，用于实现 netty 的未来接口，因此它的实例将代理我们的消息。MyApp客户端处理程序是实现promise（作为响应未来）的净管道，.setResponseFuture（...）将代理添加到管道中。

根据频道的初始化，channelFuture。channel（）可能仍然为nullnull，给我们一个NullPointerException。因此，我们需要更改上面的代码，以便从回调中插入代理：

public MyCustomFuture send(final String msg) {
  final MyCustomFuture responseFuture = new MyCustomFuture();

  channelFuture.addListener(new GenericFutureListener<ChannelFuture>() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
      channelFuture.channel().pipeline()
          .get(MyAppClientHandler.class).setResponseFuture(responseFuture);
      channelFuture.channel().writeAndFlush(msg);                               
    }
  });

  return responseFuture;
}

关于< code>MyCustomFuture的另一件事是，它需要一个setter方法:

public void set(String msg) throws InterruptedException {
  if (state == State.DONE) {
    return;
  }
  blockingReplyHolder.put(msg);
  state = State.DONE;
}

顾名思义，< code>blockingReplyHolder是实现的字段，它保存实现promise的消息，如果它仍然不存在，就进行阻塞(检查未来)

好的。现在，当预期的消息到达管道MyAppClientHandler时，我们可以实现如下promise：

protected void channelRead(ChannelHandlerContext ctx, String msg) throws Exception {
    responseFuture.set(msg);
}

生成的自定义API的用法为：

MyCustomFuture future = ioClient.send(message);
// do other stuff if needed
String response = future.get(); // waits if necessary
// make use of the response

这个答案来源于我正在玩弄的一个例子。