Building on Akka I/O

Using Akka Asynchronous Actor Networking

By Dennis M. Sosnoski, ScalableScala.com / Sosnoski Software Associates Ltd

Akka UDP server

UDP socket server example:

• Bind to listen to a local socket
• Echo any received datagrams back to sender

class UdpEchoService(address: InetSocketAddress) extends Actor {
  import context.system
  IO(Udp) ! Udp.Bind(self, address)

  def receive = {
    case Udp.Bound(_) =>
      context.become(ready(sender))
  }

  def ready(socket: ActorRef): Receive = {
    case Udp.Received(data, remote) =>
      socket ! Udp.Send(data, remote)
    case Udp.Unbind => socket ! Udp.Unbind
    case Udp.Unbound => context.stop(self)
  }
}

Akka UDP client

UDP socket client example:

• "Bind" to a local socket
• Send string messages as UDP to remote address
• Print received UDP messages to console

class UdpEchoClient(local: InetSocketAddress, remote: InetSocketAddress) extends Actor {
  import context.system
  IO(Udp) ! Udp.Bind(self, local)

  def receive = {
    case Udp.Bound(_) =>
      context.become(ready(sender))
  }

  def ready(socket: ActorRef): Receive = {
    case msg: String =>
      socket ! Udp.Send(ByteString(msg, "UTF-8"), remote)
    case Udp.Received(data, _) =>
      println(s"Client received ${data.decodeString("UTF-8")}")
    case Udp.Unbind => socket ! Udp.Unbind
    case Udp.Unbound => context.stop(self)
  }
}

SMTP state diagrams

Spray HTTP server example

class EchoHttpService(host: String, port: Int) extends Actor {
  import context.system
  IO(Http) ! Http.Bind(self, interface = host, port = port)

  def receive: Receive = {
    case Http.Connected(remote, _) =>
      sender ! Http.Register(context.actorOf(EchoHttpConnectionHandler.props(remote, sender)))
  }
}

object EchoHttpConnectionHandler {
  def props(remote: InetSocketAddress, connection: ActorRef): Props =
    Props(new EchoHttpConnectionHandler(remote, connection))
}

class EchoHttpConnectionHandler(remote: InetSocketAddress, connection: ActorRef) extends Actor {
  context.watch(connection)

  def receive: Receive = {
    case HttpRequest(HttpMethods.GET, uri, _, _, _) =>
      sender ! HttpResponse(entity = uri.path.toString)
    case _: Tcp.ConnectionClosed =>
      context.stop(self)
    case Terminated(`connection`) =>
      context.stop(self)
  }
}