Akka/Scala example source code file (IODocSpec.scala)

This example Akka source code file (IODocSpec.scala) is included in my "Source Code Warehouse" project. The intent of this project is to help you more easily find Akka and Scala source code examples by using tags.

All credit for the original source code belongs to akka.io; I'm just trying to make examples easier to find. (For my Scala work, see my Scala examples and tutorials.)

Akka tags/keywords

actor, akka, bind, bytestring, client, commandfailed, concurrent, connected, duration, inetsocketaddress, io, received, test, testing, time, write

The IODocSpec.scala Akka example source code

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

package docs.io

//#imports
import akka.actor.{ Actor, ActorRef, Props }
import akka.io.{ IO, Tcp }
import akka.util.ByteString
import java.net.InetSocketAddress
//#imports

import akka.testkit.AkkaSpec
import scala.concurrent.duration._

class DemoActor extends Actor {
  //#manager
  import akka.io.{ IO, Tcp }
  import context.system // implicitly used by IO(Tcp)

  val manager = IO(Tcp)
  //#manager

  def receive = Actor.emptyBehavior
}

//#server
class Server extends Actor {

  import Tcp._
  import context.system

  IO(Tcp) ! Bind(self, new InetSocketAddress("localhost", 0))

  def receive = {
    case b @ Bound(localAddress) =>
      //#do-some-logging-or-setup
      context.parent ! b
    //#do-some-logging-or-setup

    case CommandFailed(_: Bind) => context stop self

    case c @ Connected(remote, local) =>
      //#server
      context.parent ! c
      //#server
      val handler = context.actorOf(Props[SimplisticHandler])
      val connection = sender()
      connection ! Register(handler)
  }

}
//#server

//#simplistic-handler
class SimplisticHandler extends Actor {
  import Tcp._
  def receive = {
    case Received(data) => sender() ! Write(data)
    case PeerClosed     => context stop self
  }
}
//#simplistic-handler

//#client
object Client {
  def props(remote: InetSocketAddress, replies: ActorRef) =
    Props(classOf[Client], remote, replies)
}

class Client(remote: InetSocketAddress, listener: ActorRef) extends Actor {

  import Tcp._
  import context.system

  IO(Tcp) ! Connect(remote)

  def receive = {
    case CommandFailed(_: Connect) =>
      listener ! "connect failed"
      context stop self

    case c @ Connected(remote, local) =>
      listener ! c
      val connection = sender()
      connection ! Register(self)
      context become {
        case data: ByteString =>
          connection ! Write(data)
        case CommandFailed(w: Write) =>
          // O/S buffer was full
          listener ! "write failed"
        case Received(data) =>
          listener ! data
        case "close" =>
          connection ! Close
        case _: ConnectionClosed =>
          listener ! "connection closed"
          context stop self
      }
  }
}
//#client

class IODocSpec extends AkkaSpec {

  class Parent extends Actor {
    context.actorOf(Props[Server], "server")
    def receive = {
      case msg => testActor forward msg
    }
  }

  "demonstrate connect" in {
    val server = system.actorOf(Props(classOf[Parent], this), "parent")
    val listen = expectMsgType[Tcp.Bound].localAddress
    val client = system.actorOf(Client.props(listen, testActor), "client1")

    watch(client)

    val c1, c2 = expectMsgType[Tcp.Connected]
    c1.localAddress should be(c2.remoteAddress)
    c2.localAddress should be(c1.remoteAddress)

    client ! ByteString("hello")
    expectMsgType[ByteString].utf8String should be("hello")

    client ! "close"
    expectMsg("connection closed")
    expectTerminated(client, 1.second)
  }

}