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

This example Akka source code file (UdpDocSpec.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, actorref, akka, connected, inetsocketaddress, io, listener, props, receive, simplesender, test, testing, testkit, testprobe, udpdocspec

The UdpDocSpec.scala Akka example source code

package docs.io

import akka.testkit.AkkaSpec
import akka.actor.Actor
import akka.io.IO
import akka.io.Udp
import akka.actor.ActorRef
import java.net.InetSocketAddress
import akka.util.ByteString
import akka.testkit.TestProbe
import akka.actor.Props
import scala.concurrent.duration._
import akka.actor.PoisonPill
import akka.io.UdpConnected

object ScalaUdpDocSpec {

  //#sender
  class SimpleSender(remote: InetSocketAddress) extends Actor {
    import context.system
    IO(Udp) ! Udp.SimpleSender

    def receive = {
      case Udp.SimpleSenderReady =>
        context.become(ready(sender()))
        //#sender
        sender() ! Udp.Send(ByteString("hello"), remote)
      //#sender
    }

    def ready(send: ActorRef): Receive = {
      case msg: String =>
        send ! Udp.Send(ByteString(msg), remote)
        //#sender
        if (msg == "world") send ! PoisonPill
      //#sender
    }
  }
  //#sender

  //#listener
  class Listener(nextActor: ActorRef) extends Actor {
    import context.system
    IO(Udp) ! Udp.Bind(self, new InetSocketAddress("localhost", 0))

    def receive = {
      case Udp.Bound(local) =>
        //#listener
        nextActor forward local
        //#listener
        context.become(ready(sender()))
    }

    def ready(socket: ActorRef): Receive = {
      case Udp.Received(data, remote) =>
        val processed = // parse data etc., e.g. using PipelineStage
          //#listener
          data.utf8String
        //#listener
        socket ! Udp.Send(data, remote) // example server echoes back
        nextActor ! processed
      case Udp.Unbind  => socket ! Udp.Unbind
      case Udp.Unbound => context.stop(self)
    }
  }
  //#listener

  //#connected
  class Connected(remote: InetSocketAddress) extends Actor {
    import context.system
    IO(UdpConnected) ! UdpConnected.Connect(self, remote)

    def receive = {
      case UdpConnected.Connected =>
        context.become(ready(sender()))
        //#connected
        sender() ! UdpConnected.Send(ByteString("hello"))
      //#connected
    }

    def ready(connection: ActorRef): Receive = {
      case UdpConnected.Received(data) =>
        // process data, send it on, etc.
        //#connected
        if (data.utf8String == "hello")
          connection ! UdpConnected.Send(ByteString("world"))
      //#connected
      case msg: String =>
        connection ! UdpConnected.Send(ByteString(msg))
      case d @ UdpConnected.Disconnect => connection ! d
      case UdpConnected.Disconnected   => context.stop(self)
    }
  }
  //#connected

}

abstract class UdpDocSpec extends AkkaSpec {

  def listenerProps(next: ActorRef): Props
  def simpleSenderProps(remote: InetSocketAddress): Props
  def connectedProps(remote: InetSocketAddress): Props

  "demonstrate Udp" in {
    val probe = TestProbe()
    val listen = watch(system.actorOf(listenerProps(probe.ref)))
    val local = probe.expectMsgType[InetSocketAddress]
    val listener = probe.lastSender
    val send = system.actorOf(simpleSenderProps(local))
    probe.expectMsg("hello")
    send ! "world"
    probe.expectMsg("world")
    listen ! Udp.Unbind
    expectTerminated(listen)
  }

  "demonstrate Udp suspend reading" in {
    val probe = TestProbe()
    val listen = watch(system.actorOf(listenerProps(probe.ref)))
    val local = probe.expectMsgType[InetSocketAddress]
    val listener = probe.lastSender
    listener ! Udp.SuspendReading
    Thread.sleep(1000) // no way to find out when the above is finished
    val send = system.actorOf(simpleSenderProps(local))
    probe.expectNoMsg(500.millis)
    listener ! Udp.ResumeReading
    probe.expectMsg("hello")
    send ! "world"
    probe.expectMsg("world")
    listen ! Udp.Unbind
    expectTerminated(listen)
  }

  "demonstrate UdpConnected" in {
    val probe = TestProbe()
    val listen = watch(system.actorOf(listenerProps(probe.ref)))
    val local = probe.expectMsgType[InetSocketAddress]
    val listener = probe.lastSender
    val conn = watch(system.actorOf(connectedProps(local)))
    probe.expectMsg("hello")
    probe.expectMsg("world")
    conn ! "hello"
    probe.expectMsg("hello")
    probe.expectMsg("world")
    listen ! Udp.Unbind
    expectTerminated(listen)
    conn ! UdpConnected.Disconnect
    expectTerminated(conn)
  }

}

class ScalaUdpDocSpec extends UdpDocSpec {
  import ScalaUdpDocSpec._

  override def listenerProps(next: ActorRef) = Props(new Listener(next))
  override def simpleSenderProps(remote: InetSocketAddress) = Props(new SimpleSender(remote))
  override def connectedProps(remote: InetSocketAddress) = Props(new Connected(remote))
}

class JavaUdpDocSpec extends UdpDocSpec {
  import UdpDocTest._

  override def listenerProps(next: ActorRef) = Props(new Listener(next))
  override def simpleSenderProps(remote: InetSocketAddress) = Props(new SimpleSender(remote))
  override def connectedProps(remote: InetSocketAddress) = Props(new Connected(remote))
}