title: How to implement a simple TCP protocol using Akka Streams?
body: <p>I took a stab at implementing a simple TCP-based protocol for exchanging messages with Akka Streams (see below). However, it seems like the <em>incoming</em> messages are not processed immediately; that is, in the scenario where two messages are sent one after another from the client, the first message is only printed <em>after</em> something is sent from the server:</p>
  
  <pre><code>At t=1, on [client] A is entered
  At t=2, on [client] B is entered
  At t=3, on [server] Z is entered
  At t=4, on [server] A is printed
  At t=5, on [server] Y is entered
  At t=6, on [server] B is printed
  </code></pre>
  
  <p>What I expected/want to see:</p>
  
  <pre><code>At t=1, on [client] A is entered
  At t=2, on [server] A is printed
  At t=3, on [client] B is entered
  At t=4, on [server] B is printed
  At t=5, on [server] Z is entered
  At t=6, on [client] Z is printed
  At t=7, on [server] Y is entered
  At t=8, on [client] Y is printed
  </code></pre>
  
  <p>What am I missing? Perhaps I need to, somehow, make the sinks at both ends eager? Or is each sink somehow blocked by the corresponding source (while the source is waiting for input from the command line)?</p>
  
  <pre><code>import java.nio.charset.StandardCharsets.UTF_8
  
  import akka.actor.ActorSystem
  import akka.stream.ActorMaterializer
  import akka.stream.scaladsl.{BidiFlow, Flow, Sink, Source, Tcp}
  import akka.util.ByteString
  import com.typesafe.config.ConfigFactory
  
  import scala.io.StdIn
  
  object AkkaStreamTcpChatter extends App {
    implicit val system = ActorSystem("akka-stream-tcp-chatter", ConfigFactory.defaultReference())
    implicit val materializer = ActorMaterializer()
  
    type Message = String
    val (host, port) = ("localhost", 46235)
  
    val deserialize:ByteString =&gt; Message = _.utf8String
    val serialize:Message =&gt; ByteString = message =&gt; ByteString(message getBytes UTF_8)
  
    val incoming:Flow[ByteString, Message, _] = Flow fromFunction deserialize
    val outgoing:Flow[Message, ByteString, _] = Flow fromFunction serialize
  
    val protocol = BidiFlow.fromFlows(incoming, outgoing)
  
    def prompt(s:String):Source[Message, _] = Source fromIterator {
      () =&gt; Iterator.continually(StdIn readLine s"[$s]&gt; ")
    }
  
    val print:Sink[Message, _] = Sink foreach println
  
    args.headOption foreach {
      case "server" =&gt; server()
      case "client" =&gt; client()
    }
  
    def server():Unit =
      Tcp()
        .bind(host, port)
        .runForeach { _
          .flow
          .join(protocol)
          .runWith(prompt("S"), print)
        }
  
    def client():Unit =
      Tcp()
        .outgoingConnection(host, port)
        .join(protocol)
        .runWith(prompt("C"), print)
  }
  </code></pre>