如何使用 Akka Java API 创建一个不响应的 TCP 接收器

Question

我正在使用这个 example 作为我的起点。

在那个例子中，服务器响应修改后的文本，但在我的例子中，服务器不需要响应。

查看其他类似问题，我知道我可以传递 Empty ByteString 或使用 filter(p -> false) 来不发回任何内容。但是，在那种情况下，问题是我的 whenComplete 块没有得到执行。即异常被吞噬。有没有办法避免这种情况？对此表示感谢！

connections.runForeach(connection -> {
System.out.println("New connection from: " + connection.remoteAddress());

final Flow<ByteString, ByteString, NotUsed> echo = Flow.of(ByteString.class)
.via(Framing.delimiter(ByteString.fromString("\n"), 256, FramingTruncation.DISALLOW))
.map(ByteString::utf8String)
.map(s -> s + "!!!\n")
.map(ByteString::fromString);

connection.handleWith(echo, mat);
}, mat).whenComplete((done,throwable) -> 
    { 
    //exception handling
    }
);

Answer 1

您的分析是正确的，现在您是在服务器关闭而不是每次连接时做出反应。

对完成的各个连接的反应将在传递给连接的流中完成，如下所示：

final Tcp tcp = Tcp.get(system);
tcp.bind("127.0.0.1", 6789).runForeach((connection) -> {

  final Flow<ByteString, ByteString, NotUsed> echo = Flow.of(ByteString.class)
    .via(Framing.delimiter(ByteString.fromString("\n"), 256, FramingTruncation.DISALLOW))
    .map(ByteString::utf8String)
    .map(s -> s + "!!!\n")
    .map(ByteString::fromString)
    .watchTermination((notUsed, completionStage) -> {
      completionStage.whenComplete((done, exception) -> {
        System.out.println("Connection from " + connection.remoteAddress() + " completed");
      });
      return notUsed;
    });

  connection.handleWith(echo, materializer);
}, materializer);

如何使用 Akka Java API 创建一个不响应的 TCP 接收器

How to create a TCP receiver using Akka Java API that doesn't respond

java

streaming

tcp

akka