MultipleThreadsTest.scala

package com.rightster.ingest.unit.route

import org.apache.camel.builder.RouteBuilder
import org.apache.camel.test.junit4.CamelTestSupport
import org.junit.Test

class MultipleThreadsTest extends CamelTestSupport {

  val messages = 40

  @Test def testQueue1: Unit = {

    val mock = getMockEndpoint("mock:out-1")
    mock.expectedMessageCount(4 * messages)

    for(i <- 1 to messages){
      template.asyncSendBody("seda:queue-in-1", "wow")
    }

    assertMockEndpointsSatisfied()
    assert(false)
  }

  @Test def testQueue2: Unit = {

    val mock = getMockEndpoint("mock:out-2")
    mock.expectedMessageCount(4 * messages)

    for(i <- 1 to messages){
      template.asyncSendBody("seda:queue-in-2", "wow")
    }

    assertMockEndpointsSatisfied()
    assert(false)
  }

  @Test def testQueue3: Unit = {

    val mock = getMockEndpoint("mock:out-3")
    mock.expectedMessageCount(4 * messages)

    for(i <- 1 to messages){
      template.asyncSendBody("seda:queue-in-3", "wow")
    }

    assertMockEndpointsSatisfied()
    assert(false)
  }

  override def createRouteBuilder: RouteBuilder = new RouteBuilder() {
    override def configure(): Unit = {

      val delay = 40
      val ths   = 2

      // Queue 1
      from("seda:queue-in-1")
        .split(constant("a,b,c,d").tokenize(","))
        .to("direct:queue-out-1")

      from("direct:queue-out-1")
        .delay(delay)
        .to("mock:out-1")

      // Queue 2
      from("seda:queue-in-2")
        .threads(ths)
        .split(constant("a,b,c,d").tokenize(","))
        .to("seda:queue-out-2")

      from("seda:queue-out-2")
        .delay(delay)
        .to("mock:out-2")

      // Queue 3
      from("seda:queue-in-3")
        .threads(ths)
        .split(constant("a,b,c,d").tokenize(","))
        .to("seda:queue-out-3")

      from("seda:queue-out-3")
        .threads(ths)
        .delay(delay)
        .to("mock:out-3")
    }
  }
}