raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Tickets #1924 #1925 #2383 #2387 #1927 #1926

Browse source 
Changed unlimited blocking to block up untill the replyTimeout, and added a test for it, where manual Ack is never received ticket #1926

removed unnecesary methods
This commit is contained in:
RayRoestenburg 2012-09-03 12:08:46 +02:00
parent 112e02e965
commit d0a50f66e7
47 changed files with 1056 additions and 784 deletions
Download patch file Download diff file Expand all files Collapse all files

110
akka-camel/src/test/scala/akka/camel/ConsumerIntegrationTest.scala
View file

@ -4,6 +4,7 @@
package akka.camel
import internal.ActorActivationException
import language.postfixOps
import language.existentials
@ -11,7 +12,7 @@ import akka.actor._
import org.scalatest.matchers.MustMatchers
import org.scalatest.WordSpec
import akka.camel.TestSupport._
import org.apache.camel.model.RouteDefinition
import org.apache.camel.model.{ ProcessorDefinition, RouteDefinition }
import org.apache.camel.builder.Builder
import org.apache.camel.{ FailedToCreateRouteException, CamelExecutionException }
import java.util.concurrent.{ ExecutionException, TimeUnit, TimeoutException }
@ -19,17 +20,19 @@ import akka.actor.Status.Failure
import scala.concurrent.util.duration._
import concurrent.{ ExecutionContext, Await }
import akka.testkit._
import akka.util.Timeout
class ConsumerIntegrationTest extends WordSpec with MustMatchers with NonSharedCamelSystem {
"ConsumerIntegrationTest" must {
implicit val defaultTimeout = 10.seconds
val defaultTimeoutDuration = 10 seconds
implicit val defaultTimeout = Timeout(defaultTimeoutDuration)
implicit def ec: ExecutionContext = system.dispatcher
"Consumer must throw FailedToCreateRouteException, while awaiting activation, if endpoint is invalid" in {
filterEvents(EventFilter[ActorInitializationException](occurrences = 1), EventFilter[FailedToCreateRouteException](occurrences = 1)) {
filterEvents(EventFilter[ActorActivationException](occurrences = 1)) {
val actorRef = system.actorOf(Props(new TestActor(uri = "some invalid uri")))
intercept[FailedToCreateRouteException] {
Await.result(camel.activationFutureFor(actorRef), defaultTimeout)
Await.result(camel.activationFutureFor(actorRef), defaultTimeoutDuration)
}
}
}
@ -40,7 +43,7 @@ class ConsumerIntegrationTest extends WordSpec with MustMatchers with NonSharedC
def receive = {
case m: CamelMessage ⇒ sender ! "received " + m.bodyAs[String]
}
})
}, name = "direct-a1")
camel.sendTo("direct:a1", msg = "some message") must be("received some message")
}
@ -48,17 +51,18 @@ class ConsumerIntegrationTest extends WordSpec with MustMatchers with NonSharedC
val SHORT_TIMEOUT = 10 millis
val LONG_WAIT = 200 millis
start(new Consumer {
val ref = start(new Consumer {
override def replyTimeout = SHORT_TIMEOUT
def endpointUri = "direct:a3"
def receive = { case _ ⇒ { Thread.sleep(LONG_WAIT.toMillis); sender ! "done" } }
})
}, name = "ignore-this-deadletter-timeout-consumer-reply")
val exception = intercept[CamelExecutionException] {
camel.sendTo("direct:a3", msg = "some msg 3")
}
exception.getCause.getClass must be(classOf[TimeoutException])
stop(ref)
}
"Consumer must process messages even after actor restart" in {
@ -74,90 +78,105 @@ class ConsumerIntegrationTest extends WordSpec with MustMatchers with NonSharedC
override def postRestart(reason: Throwable) {
restarted.countDown()
}
})
}, "direct-a2")
filterEvents(EventFilter[TestException](occurrences = 1)) {
consumer ! "throw"
Await.ready(restarted, defaultTimeout)
Await.ready(restarted, defaultTimeoutDuration)
camel.sendTo("direct:a2", msg = "xyz") must be("received xyz")
}
stop(consumer)
}
"Consumer must unregister itself when stopped" in {
val consumer = start(new TestActor())
Await.result(camel.activationFutureFor(consumer), defaultTimeout)
val consumer = start(new TestActor(), name = "test-actor-unregister")
Await.result(camel.activationFutureFor(consumer), defaultTimeoutDuration)
camel.routeCount must be > (0)
system.stop(consumer)
Await.result(camel.deactivationFutureFor(consumer), defaultTimeout)
Await.result(camel.deactivationFutureFor(consumer), defaultTimeoutDuration)
camel.routeCount must be(0)
}
"Consumer must register on uri passed in through constructor" in {
val consumer = start(new TestActor("direct://test"))
Await.result(camel.activationFutureFor(consumer), defaultTimeout)
val consumer = start(new TestActor("direct://test"), name = "direct-test")
Await.result(camel.activationFutureFor(consumer), defaultTimeoutDuration)
camel.routeCount must be > (0)
camel.routes.get(0).getEndpoint.getEndpointUri must be("direct://test")
system.stop(consumer)
Await.result(camel.deactivationFutureFor(consumer), defaultTimeout)
Await.result(camel.deactivationFutureFor(consumer), defaultTimeoutDuration)
camel.routeCount must be(0)
stop(consumer)
}
"Error passing consumer supports error handling through route modification" in {
start(new ErrorThrowingConsumer("direct:error-handler-test") with ErrorPassing {
override def onRouteDefinition(rd: RouteDefinition) = {
val ref = start(new ErrorThrowingConsumer("direct:error-handler-test") {
override def onRouteDefinition = (rd: RouteDefinition) ⇒ {
rd.onException(classOf[TestException]).handled(true).transform(Builder.exceptionMessage).end
}
})
}, name = "direct-error-handler-test")
filterEvents(EventFilter[TestException](occurrences = 1)) {
camel.sendTo("direct:error-handler-test", msg = "hello") must be("error: hello")
}
stop(ref)
}
"Error passing consumer supports redelivery through route modification" in {
start(new FailingOnceConsumer("direct:failing-once-concumer") with ErrorPassing {
override def onRouteDefinition(rd: RouteDefinition) = {
val ref = start(new FailingOnceConsumer("direct:failing-once-concumer") {
override def onRouteDefinition = (rd: RouteDefinition) ⇒ {
rd.onException(classOf[TestException]).maximumRedeliveries(1).end
}
})
}, name = "direct-failing-once-consumer")
filterEvents(EventFilter[TestException](occurrences = 1)) {
camel.sendTo("direct:failing-once-concumer", msg = "hello") must be("accepted: hello")
}
stop(ref)
}
"Consumer supports manual Ack" in {
start(new ManualAckConsumer() {
val ref = start(new ManualAckConsumer() {
def endpointUri = "direct:manual-ack"
def receive = { case _ ⇒ sender ! Ack }
})
camel.template.asyncSendBody("direct:manual-ack", "some message").get(defaultTimeout.toSeconds, TimeUnit.SECONDS) must be(null) //should not timeout
}, name = "direct-manual-ack-1")
camel.template.asyncSendBody("direct:manual-ack", "some message").get(defaultTimeoutDuration.toSeconds, TimeUnit.SECONDS) must be(null) //should not timeout
stop(ref)
}
"Consumer handles manual Ack failure" in {
val someException = new Exception("e1")
start(new ManualAckConsumer() {
val ref = start(new ManualAckConsumer() {
def endpointUri = "direct:manual-ack"
def receive = { case _ ⇒ sender ! Failure(someException) }
})
}, name = "direct-manual-ack-2")
intercept[ExecutionException] {
camel.template.asyncSendBody("direct:manual-ack", "some message").get(defaultTimeout.toSeconds, TimeUnit.SECONDS)
camel.template.asyncSendBody("direct:manual-ack", "some message").get(defaultTimeoutDuration.toSeconds, TimeUnit.SECONDS)
}.getCause.getCause must be(someException)
stop(ref)
}
"Consumer should time-out, if manual Ack not received within replyTimeout and should give a human readable error message" in {
start(new ManualAckConsumer() {
val ref = start(new ManualAckConsumer() {
override def replyTimeout = 10 millis
def endpointUri = "direct:manual-ack"
def receive = { case _ ⇒ }
})
}, name = "direct-manual-ack-3")
intercept[ExecutionException] {
camel.template.asyncSendBody("direct:manual-ack", "some message").get(defaultTimeout.toSeconds, TimeUnit.SECONDS)
camel.template.asyncSendBody("direct:manual-ack", "some message").get(defaultTimeoutDuration.toSeconds, TimeUnit.SECONDS)
}.getCause.getCause.getMessage must include("Failed to get Ack")
stop(ref)
}
"respond to onRouteDefinition" in {
val ref = start(new ErrorRespondingConsumer("direct:error-responding-consumer-1"), "error-responding-consumer")
filterEvents(EventFilter[TestException](occurrences = 1)) {
val response = camel.sendTo("direct:error-responding-consumer-1", "some body")
response must be("some body has an error")
}
stop(ref)
}
}
}
@ -166,6 +185,25 @@ class ErrorThrowingConsumer(override val endpointUri: String) extends Consumer {
def receive = {
case msg: CamelMessage ⇒ throw new TestException("error: %s" format msg.body)
}
override def preRestart(reason: Throwable, message: Option[Any]) {
super.preRestart(reason, message)
sender ! Failure(reason)
}
}
class ErrorRespondingConsumer(override val endpointUri: String) extends Consumer {
def receive = {
case msg: CamelMessage ⇒ throw new TestException("Error!")
}
override def onRouteDefinition = (rd: RouteDefinition) ⇒ {
// Catch TestException and handle it by returning a modified version of the in message
rd.onException(classOf[TestException]).handled(true).transform(Builder.body.append(" has an error")).end
}
final override def preRestart(reason: Throwable, message: Option[Any]) {
super.preRestart(reason, message)
sender ! Failure(reason)
}
}
class FailingOnceConsumer(override val endpointUri: String) extends Consumer {
@ -177,6 +215,11 @@ class FailingOnceConsumer(override val endpointUri: String) extends Consumer {
else
throw new TestException("rejected: %s" format msg.body)
}
final override def preRestart(reason: Throwable, message: Option[Any]) {
super.preRestart(reason, message)
sender ! Failure(reason)
}
}
class TestActor(uri: String = "file://target/abcde") extends Consumer {
@ -184,13 +227,6 @@ class TestActor(uri: String = "file://target/abcde") extends Consumer {
def receive = { case _ ⇒ /* do nothing */ }
}
trait ErrorPassing {
self: Actor ⇒
final override def preRestart(reason: Throwable, message: Option[Any]) {
sender ! Failure(reason)
}
}
trait ManualAckConsumer extends Consumer {
override def autoAck = false
}