Start actor after it can be resolved with actorFor. #2575

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

@@ -194,6 +194,7 @@ private[akka] abstract class InternalActorRef extends ActorRef with ScalaActorRe
   /*
    * Actor life-cycle management, invoked only internally (in response to user requests via ActorContext).
    */
+  def start(): ActorRef
   def resume(causedByFailure: Throwable): Unit
   def suspend(): Unit
   def restart(cause: Throwable): Unit
@@ -265,11 +266,15 @@ private[akka] class LocalActorRef private[akka] (
    * that is reached).
    */
   private val actorCell: ActorCell = newActorCell(_system, this, _props, _supervisor)
-  actorCell.start(sendSupervise = true, ThreadLocalRandom.current.nextInt())
+  actorCell.init(ThreadLocalRandom.current.nextInt(), sendSupervise = true)
+  if (actorCellShouldStart)
+    actorCell.start()
 
   protected def newActorCell(system: ActorSystemImpl, ref: InternalActorRef, props: Props, supervisor: InternalActorRef): ActorCell =
     new ActorCell(system, ref, props, supervisor)
 
+  protected def actorCellShouldStart(): Boolean = false
+
   protected def actorContext: ActorContext = actorCell
 
   /**
@@ -279,6 +284,11 @@ private[akka] class LocalActorRef private[akka] (
    */
   override def isTerminated: Boolean = actorCell.isTerminated
 
+  override def start(): ActorRef = {
+    actorCell.start()
+    this
+  }
+
   /**
    * Suspends the actor so that it will not process messages until resumed. The
    * suspend request is processed asynchronously to the caller of this method
@@ -390,6 +400,7 @@ private[akka] trait MinimalActorRef extends InternalActorRef with LocalRef {
   override def getParent: InternalActorRef = Nobody
   override def getChild(names: Iterator[String]): InternalActorRef = if (names.forall(_.isEmpty)) this else Nobody
 
+  override def start(): ActorRef = this
   override def suspend(): Unit = ()
   override def resume(causedByFailure: Throwable): Unit = ()
   override def stop(): Unit = ()

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

@@ -516,6 +516,7 @@ class LocalActorRefProvider(
     cell.reserveChild("user")
     val ref = new LocalActorRef(system, Props(new Guardian(guardianStrategy)), rootGuardian, rootPath / "user")
     cell.initChild(ref)
+    ref.start()
     ref
   }
 
@@ -524,6 +525,7 @@ class LocalActorRefProvider(
     cell.reserveChild("system")
     val ref = new LocalActorRef(system, Props(new SystemGuardian(systemGuardianStrategy)), rootGuardian, rootPath / "system")
     cell.initChild(ref)
+    ref.start()
     ref
   }
 

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

@@ -79,7 +79,9 @@ private[akka] class RepointableActorRef(
    */
   def newCell(old: Cell): Cell =
     new ActorCell(system, this, props, supervisor)
-      .start(sendSupervise = false, old.asInstanceOf[UnstartedCell].uid)
+      .init(old.asInstanceOf[UnstartedCell].uid, sendSupervise = false).start()
+
+  def start(): ActorRef = this
 
   def suspend(): Unit = underlying.suspend()
 

akka-actor/src/main/scala/akka/actor/dungeon/Children.scala

@@ -192,7 +192,7 @@ private[akka] trait Children { this: ActorCell ⇒
       // mailbox==null during RoutedActorCell constructor, where suspends are queued otherwise
       if (mailbox ne null) for (_ ← 1 to mailbox.suspendCount) actor.suspend()
       initChild(actor)
-      actor
+      actor.start()
     }
   }
 

akka-actor/src/main/scala/akka/actor/dungeon/Dispatch.scala

@@ -38,12 +38,11 @@ private[akka] trait Dispatch { this: ActorCell ⇒
   final def isTerminated: Boolean = mailbox.isClosed
 
   /**
-   * Start this cell, i.e. attach it to the dispatcher. The UID must reasonably
-   * be different from the previous UID of a possible actor with the same path,
+   * Initialize this cell, i.e. set up mailboxes and supervision. The UID must be
+   * reasonably different from the previous UID of a possible actor with the same path,
    * which can be achieved by using ThreadLocalRandom.current.nextInt().
    */
-  final def start(sendSupervise: Boolean, uid: Int): this.type = {
-
+  final def init(uid: Int, sendSupervise: Boolean): this.type = {
     /*
      * Create the mailbox and enqueue the Create() message to ensure that
      * this is processed before anything else.
@@ -59,10 +58,15 @@ private[akka] trait Dispatch { this: ActorCell ⇒
       parent.sendSystemMessage(akka.dispatch.Supervise(self, uid))
       parent ! NullMessage // read ScalaDoc of NullMessage to see why
     }
+    this
+  }
 
+  /**
+   * Start this cell, i.e. attach it to the dispatcher.
+   */
+  final def start(): this.type = {
     // This call is expected to start off the actor by scheduling its mailbox.
     dispatcher.attach(this)
-
     this
   }
 

akka-actor/src/main/scala/akka/routing/Routing.scala

@@ -71,7 +71,7 @@ private[akka] class RoutedActorCell(_system: ActorSystemImpl, _ref: InternalActo
     r
   }
 
-  start(sendSupervise = false, _uid)
+  init(_uid, sendSupervise = false).start()
 
   /*
    * end of construction

akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala

@@ -243,6 +243,8 @@ private[akka] class RemoteActorRef private[akka] (
         provider.deadLetters ! message
     }
 
+  def start(): ActorRef = this
+
   def suspend(): Unit = sendSystemMessage(Suspend())
 
   def resume(causedByFailure: Throwable): Unit = sendSystemMessage(Resume(causedByFailure))

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

@@ -54,6 +54,9 @@ class TestActorRef[T <: Actor](
       }
     }
 
+  // we need to start ourselves since the creation of an actor has been split into initialization and starting
+  override def actorCellShouldStart(): Boolean = true
+
   /**
    * Directly inject messages into actor receive behavior. Any exceptions
    * thrown will be available to you, while still being able to use