Akka issue demo

Test.scala

import akka.actor.ActorSystem
import akka.stream.scaladsl.{ Flow, Framing, Source, Tcp }
import akka.util.ByteString

import scala.concurrent.{ ExecutionContext, Future }

object Test {

  def logic(s: String)(implicit ec: ExecutionContext): Future[String] = {
    Future {
      val r = s.reverse
      if (s == r) throw new IllegalArgumentException("S = R!")
      r
    }
  }

  def main(args: Array[String]): Unit = {
    println("Let's go")

    implicit val system: ActorSystem = ActorSystem("Test-me")
    import system.dispatcher

    val serverConn = Tcp().bind("localhost", 11111)

    serverConn runForeach { conn =>

      val handler = Flow[ByteString]
        .via(Framing.delimiter(ByteString("\r\n"), 128))
        .map(_.utf8String)
        .mapAsync(1)(logic)
        .map(s => Seq(s + "\n", "-----------------\n"))
        .flatMapConcat { rs =>
          val it = rs.map { resp =>
            ByteString(resp)
          }
          Source(it.toList)
        }
        .recover { case x => ByteString(s"ERROR: ${x.getMessage}\n") }

      conn.handleWith(handler)
    }
  }
}