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

This example Akka source code file (TestTransportSpec.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

a, address, akka, associationregistry, concurrent, defaulttimeout, expect, future, inboundassociation, inboundpayload, remote, test, testing, testtransport, transport, writeattempt

The TestTransportSpec.scala Akka example source code

package akka.remote.transport

import akka.testkit._
import scala.concurrent._
import akka.actor.Address
import akka.remote.transport.Transport._
import akka.remote.transport.TestTransport._
import akka.util.ByteString
import akka.remote.transport.AssociationHandle.{ ActorHandleEventListener, Disassociated, InboundPayload }

class TestTransportSpec extends AkkaSpec with DefaultTimeout with ImplicitSender {

  val addressA: Address = Address("test", "testsytemA", "testhostA", 4321)
  val addressB: Address = Address("test", "testsytemB", "testhostB", 5432)
  val nonExistingAddress = Address("test", "nosystem", "nohost", 0)

  "TestTransport" must {

    "return an Address and promise when listen is called and log calls" in {
      val registry = new AssociationRegistry
      val transportA = new TestTransport(addressA, registry)

      val result = Await.result(transportA.listen, timeout.duration)

      result._1 should be(addressA)
      result._2 should not be null

      registry.logSnapshot.exists {
        case ListenAttempt(address) ⇒ address == addressA
        case _                      ⇒ false
      } should be(true)
    }

    "associate successfully with another TestTransport and log" in {
      val registry = new AssociationRegistry
      val transportA = new TestTransport(addressA, registry)
      val transportB = new TestTransport(addressB, registry)

      // Must complete the returned promise to receive events
      Await.result(transportA.listen, timeout.duration)._2.success(ActorAssociationEventListener(self))
      Await.result(transportB.listen, timeout.duration)._2.success(ActorAssociationEventListener(self))

      awaitCond(registry.transportsReady(addressA, addressB))

      transportA.associate(addressB)
      expectMsgPF(timeout.duration, "Expect InboundAssociation from A") {
        case InboundAssociation(handle) if handle.remoteAddress == addressA ⇒
      }

      registry.logSnapshot.contains(AssociateAttempt(addressA, addressB)) should be(true)
    }

    "fail to associate with nonexisting address" in {
      val registry = new AssociationRegistry
      var transportA = new TestTransport(addressA, registry)

      Await.result(transportA.listen, timeout.duration)._2.success(ActorAssociationEventListener(self))

      // TestTransport throws IllegalAssociationException when trying to associate with non-existing system
      intercept[InvalidAssociationException] { Await.result(transportA.associate(nonExistingAddress), timeout.duration) }

    }

    "emulate sending PDUs and logs write" in {
      val registry = new AssociationRegistry
      val transportA = new TestTransport(addressA, registry)
      val transportB = new TestTransport(addressB, registry)

      Await.result(transportA.listen, timeout.duration)._2.success(ActorAssociationEventListener(self))
      Await.result(transportB.listen, timeout.duration)._2.success(ActorAssociationEventListener(self))

      awaitCond(registry.transportsReady(addressA, addressB))

      val associate: Future[AssociationHandle] = transportA.associate(addressB)
      val handleB = expectMsgPF(timeout.duration, "Expect InboundAssociation from A") {
        case InboundAssociation(handle) if handle.remoteAddress == addressA ⇒ handle
      }

      handleB.readHandlerPromise.success(ActorHandleEventListener(self))
      val handleA = Await.result(associate, timeout.duration)

      // Initialize handles
      handleA.readHandlerPromise.success(ActorHandleEventListener(self))

      val akkaPDU = ByteString("AkkaPDU")

      awaitCond(registry.existsAssociation(addressA, addressB))

      handleA.write(akkaPDU)
      expectMsgPF(timeout.duration, "Expect InboundPayload from A") {
        case InboundPayload(payload) if payload == akkaPDU ⇒
      }

      registry.logSnapshot.exists {
        case WriteAttempt(sender, recipient, payload) ⇒
          sender == addressA && recipient == addressB && payload == akkaPDU
        case _ ⇒ false
      } should be(true)
    }

    "emulate disassociation and log it" in {
      val registry = new AssociationRegistry
      val transportA = new TestTransport(addressA, registry)
      val transportB = new TestTransport(addressB, registry)

      Await.result(transportA.listen, timeout.duration)._2.success(ActorAssociationEventListener(self))
      Await.result(transportB.listen, timeout.duration)._2.success(ActorAssociationEventListener(self))

      awaitCond(registry.transportsReady(addressA, addressB))

      val associate: Future[AssociationHandle] = transportA.associate(addressB)
      val handleB: AssociationHandle = expectMsgPF(timeout.duration, "Expect InboundAssociation from A") {
        case InboundAssociation(handle) if handle.remoteAddress == addressA ⇒ handle
      }

      handleB.readHandlerPromise.success(ActorHandleEventListener(self))
      val handleA = Await.result(associate, timeout.duration)

      // Initialize handles
      handleA.readHandlerPromise.success(ActorHandleEventListener(self))

      awaitCond(registry.existsAssociation(addressA, addressB))

      handleA.disassociate()

      expectMsgPF(timeout.duration) {
        case Disassociated(_) ⇒
      }

      awaitCond(!registry.existsAssociation(addressA, addressB))

      registry.logSnapshot exists {
        case DisassociateAttempt(requester, remote) if requester == addressA && remote == addressB ⇒ true
        case _ ⇒ false
      } should be(true)
    }

  }

}