content.scala

/**
 * Copyright (C) 2014 Typesafe Inc. <http://www.typesafe.com>
 */
package example

import java.net.InetSocketAddress

import akka.actor.ActorSystem
import akka.pattern.FutureTimeoutSupport
import akka.stream.FlowMaterializer
import akka.stream.scaladsl.Flow
import akka.stream.scaladsl.Source
import akka.stream.scaladsl.StreamTcp
import akka.stream.scaladsl.StreamTcp.IncomingConnection
import akka.stream.scaladsl.StreamTcp.OutgoingConnection
import akka.util.ByteString

object IO extends App with FutureTimeoutSupport {
  implicit val sys = ActorSystem()
  implicit val mat = FlowMaterializer()

  private val localhost = new InetSocketAddress("127.0.0.1", 8888)
  val binding = StreamTcp().bind(localhost)
  val connections: Source[IncomingConnection] = binding.connections

  connections foreach { connection ⇒
    println(s"Connection from: ${connection.remoteAddress}")
    connection.handleWith(Flow[ByteString].map(_.dropRight(1) ++ ByteString("!!!")))
  }

  val connection: OutgoingConnection = StreamTcp().outgoingConnection(localhost)

  readLine("waiting...")
  sys.shutdown()
}

object Writer extends App with FutureTimeoutSupport {
  implicit val sys = ActorSystem()
  implicit val mat = FlowMaterializer()

  private val localhost = new InetSocketAddress("127.0.0.1", 8888)
  val binding = StreamTcp().outgoingConnection(localhost)

  val repl = Flow[ByteString]
    .map(s ⇒ println("Server: " + s))
    .map(_ ⇒ readLine("> "))
    .map {
      case "q" ⇒
        sys.shutdown(); ByteString("")
      case text ⇒ ByteString(text)
    }

  binding.handleWith(repl)
}