Allow single connection with Akka stream as tcp server

Question

I am trying to build a simple tcp server using Akka streams.

 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)
  }

I want to allow maximum one connection at a time. To achieve this I have added the following line into my application.conf file.

akka.io.tcp.max-channels = 2

With this configuration, akka allows only one connection at a time. However, as soon as the second connection is attempted it rejects the request and fails itself with the following message:

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

From this point, it is not possible to establish any connection since the Tcp server is down.

Question: What is the proper way of enabling only one connection at a time? The main purpose is answering the first connection request and rejecting others while it is still in progress. It should be again possible to make another connection, after the previous connection is closed. As I mentioned, only one connection should be allowed at any time.

BONUS: Is it possible to provide a whitelist to make akka stream accept connections only from this list? I am planning to allow only known ip addresses to connect my server. To achieve this, I think it is enough to know the proper way of rejecting a request. So I can compare the incoming connection's ip address with a given list and reject if it is not in there. But any better solution is also appreciated.

What do you want to do with those connection that come after the processing one? If your goal is just to reject upcoming connections, the simple way I see it is to introduce some kind of semaphore that indicates the number of requests processes simultaneously, but instead of blocking, just reject new connections — maks
Regarding BONUS: do you want such whitelist on network or application level? — maks
Have not worked with streams, but as hint could say that after the bind you get Source Tcp.IncomingConnection which have a possibility to get the address. After that you could in some way reject that connection, but it depends what is rejection in your terms. It could be just an another handler that writes some error data to the incoming connection — maks
@maks, I am kind of new on tcp connections. But as a first option, (if possible) I would like to close the connection and not to allow the client to send anything to server. Your recommendation works as well, but the question remains: what is the proper way to implement any of them? — fcat

maks maks · Accepted Answer · 2016-12-28T10:12:55

Bind method of Tcp has a parameter options which accepts a Traversable of Socket options. You can pass smth like this to that paramter:

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

so your code will look like this:

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)
  }

The approach of limiting the number of request is the same, investigate methods in the SocketOptions trait

PS. Have not tried this to run, just concluded after investigation of stream API, so please check for correctness.

Allow single connection with Akka stream as tcp server

1 Answers