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

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

activationaware, activationintegrationtest, akka, camelmessage, concurrent, consumer, duration, matchers, sharedcamelsystem, string, test, testconsumer, testing, testlatch, time, timeout

The ActivationIntegrationTest.scala Akka example source code

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

package akka.camel

import language.postfixOps

import org.scalatest.Matchers
import scala.concurrent.duration._
import org.apache.camel.ProducerTemplate
import akka.actor._
import TestSupport._
import org.scalatest.WordSpec
import akka.testkit.TestLatch
import scala.concurrent.Await
import java.util.concurrent.TimeoutException
import akka.util.Timeout

class ActivationIntegrationTest extends WordSpec with Matchers with SharedCamelSystem {
  val timeoutDuration = 10 seconds
  implicit val timeout = Timeout(timeoutDuration)
  def template: ProducerTemplate = camel.template
  import system.dispatcher

  "ActivationAware should be notified when endpoint is activated" in {
    val latch = new TestLatch(0)
    val actor = system.actorOf(Props(new TestConsumer("direct:actor-1", latch)), "act-direct-actor-1")
    Await.result(camel.activationFutureFor(actor), 10 seconds) should be(actor)

    template.requestBody("direct:actor-1", "test") should be("received test")
  }

  "ActivationAware should be notified when endpoint is de-activated" in {
    val latch = TestLatch(1)
    val actor = start(new Consumer {
      def endpointUri = "direct:a3"
      def receive = { case _ ⇒ {} }

      override def postStop() {
        super.postStop()
        latch.countDown()
      }
    }, name = "direct-a3")
    Await.result(camel.activationFutureFor(actor), timeoutDuration)

    system.stop(actor)
    Await.result(camel.deactivationFutureFor(actor), timeoutDuration)
    Await.ready(latch, timeoutDuration)
  }

  "ActivationAware must time out when waiting for endpoint de-activation for too long" in {
    val latch = new TestLatch(0)
    val actor = start(new TestConsumer("direct:a5", latch), name = "direct-a5")
    Await.result(camel.activationFutureFor(actor), timeoutDuration)
    intercept[TimeoutException] { Await.result(camel.deactivationFutureFor(actor), 1 millis) }
  }

  "activationFutureFor must fail if notification timeout is too short and activation is not complete yet" in {
    val latch = new TestLatch(1)
    val actor = system.actorOf(Props(new TestConsumer("direct:actor-4", latch)), "direct-actor-4")
    intercept[TimeoutException] { Await.result(camel.activationFutureFor(actor), 1 millis) }
    latch.countDown()
    // after the latch is removed, complete the wait for completion so this test does not later on
    // print errors because of the registerConsumer timing out.
    Await.result(camel.activationFutureFor(actor), timeoutDuration)
  }

  class TestConsumer(uri: String, latch: TestLatch) extends Consumer {
    def endpointUri = uri

    override def preStart() {
      Await.ready(latch, 60 seconds)
      super.preStart()
    }

    override def receive = {
      case msg: CamelMessage ⇒ sender() ! "received " + msg.body
    }
  }

}