Less confusing logging for dead letters without sender (akka#22722)

Of course it'd be much neater if `DeadLetter.sender` were an `Option` to begin
with, but we can't do that due to backwards compatibility.

akka-actor-tests/src/test/scala/akka/actor/ActorSystemSpec.scala

@@ -157,8 +157,27 @@ class ActorSystemSpec extends AkkaSpec(ActorSystemSpec.config) with ImplicitSend
         probe.watch(a)
         a.tell("run", probe.ref)
         probe.expectTerminated(a)
-        EventFilter.info(pattern = "not delivered", occurrences = 1).intercept {
-          a.tell("boom", probe.ref)
+        EventFilter.info(pattern = """from Actor\[akka://LogDeadLetters/system/testProbe.*not delivered""", occurrences = 1).intercept {
+          EventFilter.warning(pattern = """received dead letter from Actor\[akka://LogDeadLetters/system/testProbe""", occurrences = 1).intercept {
+            a.tell("boom", probe.ref)
+          }(sys)
+        }(sys)
+
+      } finally shutdown(sys)
+    }
+
+    "log dead letters sent without sender reference" in {
+      val sys = ActorSystem("LogDeadLetters", ConfigFactory.parseString("akka.loglevel=INFO").withFallback(AkkaSpec.testConf))
+      try {
+        val probe = TestProbe()(sys)
+        val a = sys.actorOf(Props[ActorSystemSpec.Terminater])
+        probe.watch(a)
+        a.tell("run", probe.ref)
+        probe.expectTerminated(a)
+        EventFilter.info(pattern = "without sender.*not delivered", occurrences = 1).intercept {
+          EventFilter.warning(pattern = "received dead letter without sender", occurrences = 1).intercept {
+            a.tell("boom", ActorRef.noSender)
+          }(sys)
         }(sys)
 
       } finally shutdown(sys)

akka-actor-tests/src/test/scala/akka/dispatch/MailboxConfigSpec.scala

@@ -134,7 +134,7 @@ abstract class MailboxSpec extends AkkaSpec with BeforeAndAfterAll with BeforeAn
     ensureInitialMailboxState(config, q)
 
     EventFilter.warning(
-      pattern = ".*received dead letter from Actor.*MailboxSpec/deadLetters.*",
+      pattern = "received dead letter without sender",
       occurrences = (enqueueN - dequeueN)) intercept {
 
       def createProducer(fromNum: Int, toNum: Int): Future[Vector[Envelope]] = spawn {

akka-actor/src/main/scala/akka/actor/ActorRef.scala

@@ -472,7 +472,9 @@ sealed trait AllDeadLetters {
 
 /**
  * When a message is sent to an Actor that is terminated before receiving the message, it will be sent as a DeadLetter
- * to the ActorSystem's EventStream
+ * to the ActorSystem's EventStream.
+ *
+ * When this message was sent without a sender [[ActorRef]], `sender` will be `system.deadLetters`.
  */
 @SerialVersionUID(1L)
 final case class DeadLetter(message: Any, sender: ActorRef, recipient: ActorRef) extends AllDeadLetters {

akka-actor/src/main/scala/akka/event/DeadLetterListener.scala

@@ -28,10 +28,11 @@ class DeadLetterListener extends Actor {
   def receive = {
     case DeadLetter(message, snd, rcp) ⇒
       count += 1
+      val origin = if (snd eq context.system.deadLetters) "without sender" else s"from $snd"
       val done = maxCount != Int.MaxValue && count >= maxCount
       val doneMsg = if (done) ", no more dead letters will be logged" else ""
       eventStream.publish(Info(rcp.path.toString, rcp.getClass,
-        s"Message [${message.getClass.getName}] from $snd to $rcp was not delivered. [$count] dead letters encountered$doneMsg. " +
+        s"Message [${message.getClass.getName}] $origin to $rcp was not delivered. [$count] dead letters encountered$doneMsg. " +
           "This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' " +
           "and 'akka.log-dead-letters-during-shutdown'."))
       if (done) context.stop(self)

akka-testkit/src/main/scala/akka/testkit/TestEventListener.scala

@@ -12,7 +12,7 @@ import akka.actor.{ DeadLetter, ActorSystem, UnhandledMessage }
 import akka.dispatch.sysmsg.{ SystemMessage, Terminate }
 import akka.event.Logging.{ Warning, LogEvent, InitializeLogger, Info, Error, Debug, LoggerInitialized }
 import akka.event.Logging
-import akka.actor.NoSerializationVerificationNeeded
+import akka.actor.{ ActorRef, NoSerializationVerificationNeeded }
 import akka.japi.Util.immutableSeq
 import java.lang.{ Iterable ⇒ JIterable }
 import akka.util.BoxedType
@@ -508,10 +508,11 @@ class TestEventListener extends Logging.DefaultLogger {
       if (!msg.isInstanceOf[Terminate]) {
         val event = Warning(rcp.path.toString, rcp.getClass, msg)
         if (!filter(event)) {
-          val msgStr =
-            if (msg.isInstanceOf[SystemMessage]) "received dead system message: " + msg
-            else "received dead letter from " + snd + ": " + msg
-          val event2 = Warning(rcp.path.toString, rcp.getClass, msgStr)
+          val msgPrefix =
+            if (msg.isInstanceOf[SystemMessage]) "received dead system message"
+            else if (snd eq context.system.deadLetters) "received dead letter without sender"
+            else "received dead letter from " + snd
+          val event2 = Warning(rcp.path.toString, rcp.getClass, msgPrefix + ": " + msg)
           if (!filter(event2)) print(event2)
         }
       }