RecipientList.scala

import org.scalatest.matchers.MustMatchers

import scalaz._
import scalaz.camel._

object RouterConfig {
  import org.apache.camel.impl.DefaultCamelContext

  val context = new DefaultCamelContext
  val template = context.createProducerTemplate

  implicit val router = new Router(context)
}

object RecipientListDemo extends Application with MustMatchers {
  import concurrent.Strategy
  import Scalaz._
  import Camel._
  import RouterConfig._

  router.start

  // some message processors
  def append(s: String) = (msg: Message) => msg.appendBody(s)
  def delay(ms: Long) = (msg: Message) => { Thread sleep ms; msg }

  // function to combine results from independent computations
  val aggregator = (m1: Message, m2: Message) => m1.appendBody(" - %s" format m2.body)

  // creates an example route consuming from endpoint defined by uri
  def route(uri: String)(implicit s: Strategy) =
    from(uri) route {
      append("-begin") >=> multicast (
        delay(3000) >=> append("-1"),
        delay(3000) >=> append("-2"),
        delay(3000) >=> append("-3")
      ) (aggregator) >=> append("-end")
    }

  {
    // strategy causing a parallel multicast where each
    // computation is run in a separate, newly created
    // thread
    import concurrent.Strategy.Naive

    // create route consuming from direct:test-1
    route("direct:test-1")

    // takes about 3 seconds to complete (parallel multicast)
    template.requestBody("direct:test-1", "test") must
      equal("test-begin-1 - test-begin-2 - test-begin-3-end")
  }

  {
    // strategy causing a sequential multicast where
    // each computation is run in the main thread
    import concurrent.Strategy.Sequential

    // create route consuming from direct:test-2
    route("direct:test-2")

    // takes about 9 (3x3) seconds to complete (sequential multicast)
    template.requestBody("direct:test-2", "test") must
      equal("test-begin-1 - test-begin-2 - test-begin-3-end")
  }

  router.stop
}