scala - 允许与 Akka 流作为 tcp 服务器的单一连接

Question

我正在尝试使用 Akka 流构建一个简单的 tcp 服务器。

 Tcp()
  .bind(props.host, props.port)
  .to(Sink.foreach(_.handleWith(handler)))
  .run()
  .onComplete {
    case Success(i) => logger.info(s"Server is bound at ${props.host}:${props.port}")
    case Failure(e) => logger.error("Server binding failure", e)
  }

我想一次允许最多一个连接。为此，我将以下行添加到我的 application.conf 文件中。

akka.io.tcp.max-channels = 2

通过这种配置，akka 一次只允许一个连接。但是，一旦尝试第二次连接，它就会拒绝请求并失败并显示以下消息:

Could not register incoming connection since selector capacity limit is reached, closing connection

从这一点来看，不可能建立任何连接，因为 Tcp 服务器已关闭。

问题:一次只启用一个连接的正确方法是什么？主要目的是回答第一个连接请求并拒绝其他仍在进行中的连接请求。在关闭前一个连接后，应该可以再次建立另一个连接。正如我所提到的，任何时候都应该只允许一个连接。

奖励:是否可以提供白名单以使 akka 流仅接受来自此列表的连接？ 我打算只允许已知的 IP 地址连接我的服务器。要做到这一点，我认为知道拒绝请求的正确方法就足够了。所以我可以将传入连接的 IP 地址与给定列表进行比较，如果不存在则拒绝。但也欢迎任何更好的解决方案。

Answer 1

Tcp 的 Bind 方法有一个参数 options，它接受一个 Traversable of Socket 选项。您可以像这样将 smth 传递给该参数:

case class AllowedAddresses(addresses: Seq[InetAddress]) extends SocketOption {
    override def beforeConnect(s: Socket): Unit = {
      if (!addresses.contains(s.getInetAddress)) s.close()
    }
  }

因此您的代码将如下所示:

Tcp()
  .bind(props.host, props.port, options = List(AllowedAddresses(listOfAddresses)))
  .to(Sink.foreach(_.handleWith(handler)))
  .run()
  .onComplete {
    case Success(i) => logger.info(s"Server is bound at ${props.host}:${props.port}")
    case Failure(e) => logger.error("Server binding failure", e)
  }

限制请求数的方法是一样的，查看SocketOptions trait中的方法

附言。没试过运行，查过stream API才得出结论，请检查正确性。