From 18260a3b7bcda1055572eef472cb654d62c25604 Mon Sep 17 00:00:00 2001 From: Viktor Klang Date: Fri, 8 Jun 2012 13:56:53 +0200 Subject: [PATCH] #2203 - publish failed message deliveries to DeadLetters when bounded or durable mailbox enqueues fail --- .../actor/ActorWithBoundedStashSpec.scala | 51 ++++++------------- .../akka/dispatch/MailboxConfigSpec.scala | 10 ++-- .../main/scala/akka/dispatch/Mailbox.scala | 35 +++++-------- 3 files changed, 32 insertions(+), 64 deletions(-) diff --git a/akka-actor-tests/src/test/scala/akka/actor/ActorWithBoundedStashSpec.scala b/akka-actor-tests/src/test/scala/akka/actor/ActorWithBoundedStashSpec.scala index 9d411268aa..33283b18cf 100644 --- a/akka-actor-tests/src/test/scala/akka/actor/ActorWithBoundedStashSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/actor/ActorWithBoundedStashSpec.scala @@ -6,7 +6,7 @@ package akka.actor import akka.testkit._ import akka.testkit.DefaultTimeout import akka.testkit.TestEvent._ -import akka.dispatch.{ Await, MessageQueueAppendFailedException, BoundedDequeBasedMailbox } +import akka.dispatch.{ Await, BoundedDequeBasedMailbox } import akka.pattern.ask import akka.util.duration._ import akka.actor.ActorSystem.Settings @@ -17,16 +17,8 @@ object ActorWithBoundedStashSpec { class StashingActor(implicit sys: ActorSystem) extends Actor with Stash { def receive = { - case "hello" ⇒ - stash() - sender ! "OK" - case "world" ⇒ - try { - unstashAll() - } catch { - case e: MessageQueueAppendFailedException ⇒ - expectedException.open() - } + case "hello" ⇒ stash() + case "world" ⇒ unstashAll() } } @@ -36,18 +28,10 @@ object ActorWithBoundedStashSpec { def receive = { case "hello" ⇒ numStashed += 1 - try { - stash() - } catch { - case e: StashOverflowException ⇒ - if (numStashed == 21) stashOverflow.open() - } + try stash() catch { case e: StashOverflowException ⇒ if (numStashed == 21) sender ! "STASHOVERFLOW" } } } - @volatile var expectedException: TestLatch = null - @volatile var stashOverflow: TestLatch = null - val testConf: Config = ConfigFactory.parseString(""" my-dispatcher { mailbox-type = "akka.actor.ActorWithBoundedStashSpec$Bounded" @@ -56,47 +40,42 @@ object ActorWithBoundedStashSpec { """) // bounded deque-based mailbox with capacity 10 - class Bounded(settings: Settings, config: Config) extends BoundedDequeBasedMailbox(10, 5 seconds) - + class Bounded(settings: Settings, config: Config) extends BoundedDequeBasedMailbox(10, 1 seconds) } @org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner]) -class ActorWithBoundedStashSpec extends AkkaSpec(ActorWithBoundedStashSpec.testConf) with DefaultTimeout with BeforeAndAfterEach { +class ActorWithBoundedStashSpec extends AkkaSpec(ActorWithBoundedStashSpec.testConf) with DefaultTimeout with BeforeAndAfterEach with ImplicitSender { import ActorWithBoundedStashSpec._ implicit val sys = system - override def atStartup { - system.eventStream.publish(Mute(EventFilter[Exception]("Crashing..."))) - } + override def atStartup { system.eventStream.publish(Mute(EventFilter[Exception]("Crashing..."))) } def myProps(creator: ⇒ Actor): Props = Props(creator).withDispatcher("my-dispatcher") "An Actor with Stash and BoundedDequeBasedMailbox" must { - "throw a MessageQueueAppendFailedException in case of a capacity violation" in { - ActorWithBoundedStashSpec.expectedException = new TestLatch + "end up in DeadLetters in case of a capacity violation" in { + system.eventStream.subscribe(testActor, classOf[DeadLetter]) + val stasher = system.actorOf(myProps(new StashingActor)) // fill up stash - val futures = for (_ ← 1 to 11) yield { stasher ? "hello" } - futures foreach { Await.ready(_, 10 seconds) } + (1 to 11) foreach { _ ⇒ stasher ! "hello" } // cause unstashAll with capacity violation stasher ! "world" - Await.ready(ActorWithBoundedStashSpec.expectedException, 10 seconds) + expectMsg(DeadLetter("hello", testActor, stasher)) + system.eventStream.unsubscribe(testActor, classOf[DeadLetter]) } - } "An Actor with bounded Stash" must { "throw a StashOverflowException in case of a stash capacity violation" in { - ActorWithBoundedStashSpec.stashOverflow = new TestLatch val stasher = system.actorOf(myProps(new StashingActorWithOverflow)) // fill up stash - for (_ ← 1 to 21) { stasher ! "hello" } - Await.ready(ActorWithBoundedStashSpec.stashOverflow, 10 seconds) + (1 to 21) foreach { _ ⇒ stasher ! "hello" } + expectMsg("STASHOVERFLOW") } - } } diff --git a/akka-actor-tests/src/test/scala/akka/dispatch/MailboxConfigSpec.scala b/akka-actor-tests/src/test/scala/akka/dispatch/MailboxConfigSpec.scala index 8759f1aad9..4f2d61de65 100644 --- a/akka-actor-tests/src/test/scala/akka/dispatch/MailboxConfigSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/dispatch/MailboxConfigSpec.scala @@ -6,9 +6,8 @@ import java.util.concurrent.ConcurrentLinkedQueue import akka.util._ import akka.util.duration._ import akka.testkit.AkkaSpec -import akka.actor.{ ActorRef, ActorContext, Props, LocalActorRef } import com.typesafe.config.Config -import akka.actor.ActorSystem +import akka.actor._ @org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner]) abstract class MailboxSpec extends AkkaSpec with BeforeAndAfterAll with BeforeAndAfterEach { @@ -39,9 +38,10 @@ abstract class MailboxSpec extends AkkaSpec with BeforeAndAfterAll with BeforeAn q.numberOfMessages must be === config.capacity q.hasMessages must be === true - intercept[MessageQueueAppendFailedException] { - q.enqueue(null, exampleMessage) - } + system.eventStream.subscribe(testActor, classOf[DeadLetter]) + q.enqueue(testActor, exampleMessage) + expectMsg(DeadLetter(exampleMessage.message, system.deadLetters, testActor)) + system.eventStream.unsubscribe(testActor, classOf[DeadLetter]) q.dequeue must be === exampleMessage q.numberOfMessages must be(config.capacity - 1) diff --git a/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala b/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala index b6af478ac7..25fc0250af 100644 --- a/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala +++ b/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala @@ -6,18 +6,11 @@ package akka.dispatch import akka.AkkaException import java.util.{ Comparator, PriorityQueue, Queue, Deque } import akka.util._ -import akka.actor.{ ActorCell, ActorRef } import java.util.concurrent._ import annotation.tailrec import akka.event.Logging.Error -import akka.actor.ActorContext import com.typesafe.config.Config -import akka.actor.ActorSystem - -/** - * This exception normally is thrown when a bounded mailbox is over capacity - */ -class MessageQueueAppendFailedException(message: String, cause: Throwable = null) extends AkkaException(message, cause) +import akka.actor._ /** * INTERNAL API @@ -401,13 +394,11 @@ trait BoundedMessageQueueSemantics extends QueueBasedMessageQueue { def pushTimeOut: Duration override def queue: BlockingQueue[Envelope] - def enqueue(receiver: ActorRef, handle: Envelope) { + def enqueue(receiver: ActorRef, handle: Envelope): Unit = if (pushTimeOut.length > 0) { - queue.offer(handle, pushTimeOut.length, pushTimeOut.unit) || { - throw new MessageQueueAppendFailedException("Couldn't enqueue message " + handle + " to " + receiver) - } + if (!queue.offer(handle, pushTimeOut.length, pushTimeOut.unit)) + receiver.asInstanceOf[InternalActorRef].provider.deadLetters ! DeadLetter(handle.message, handle.sender, receiver) } else queue put handle - } def dequeue(): Envelope = queue.poll() } @@ -439,18 +430,16 @@ trait BoundedDequeBasedMessageQueueSemantics extends DequeBasedMessageQueue { override def queue: BlockingDeque[Envelope] def enqueue(receiver: ActorRef, handle: Envelope): Unit = - if (pushTimeOut.length > 0) - queue.offer(handle, pushTimeOut.length, pushTimeOut.unit) || { - throw new MessageQueueAppendFailedException("Couldn't enqueue message " + handle + " to " + receiver) - } - else queue put handle + if (pushTimeOut.length > 0) { + if (!queue.offer(handle, pushTimeOut.length, pushTimeOut.unit)) + receiver.asInstanceOf[InternalActorRef].provider.deadLetters ! DeadLetter(handle.message, handle.sender, receiver) + } else queue put handle def enqueueFirst(receiver: ActorRef, handle: Envelope): Unit = - if (pushTimeOut.length > 0) - queue.offerFirst(handle, pushTimeOut.length, pushTimeOut.unit) || { - throw new MessageQueueAppendFailedException("Couldn't enqueue message " + handle + " to " + receiver) - } - else queue putFirst handle + if (pushTimeOut.length > 0) { + if (!queue.offerFirst(handle, pushTimeOut.length, pushTimeOut.unit)) + receiver.asInstanceOf[InternalActorRef].provider.deadLetters ! DeadLetter(handle.message, handle.sender, receiver) + } else queue putFirst handle def dequeue(): Envelope = queue.poll() }