Actor Creation and Lifecycle

/**
 * Immutable configuration object for creating actors
 */
final case class Props(
  deploy: Deploy = Deploy.NoScopeGiven,
  clazz: Class[_],
  args: immutable.Seq[Any] = immutable.Seq.empty
) {
  /** Configure the dispatcher for the actor */
  def withDispatcher(dispatcher: String): Props
  
  /** Configure the mailbox for the actor */
  def withMailbox(mailbox: String): Props
  
  /** Configure the router for the actor */
  def withRouter(r: RouterConfig): Props
  
  /** Configure deployment settings */
  def withDeploy(d: Deploy): Props
}

object Props {
  /** Create Props for Java actor classes with arguments */
  def create[T <: Actor](clazz: Class[T], args: AnyRef*): Props
  
  /** Create Props with creator function (Scala API) */
  def apply[T <: Actor: ClassTag](): Props
  def apply[T <: Actor: ClassTag](creator: => T): Props
  
  /** Empty Props instance */
  val empty: Props
}

import akka.actor.{Props, Actor}

// Simple actor creation
class MyActor extends Actor {
  def receive = {
    case msg => println(s"Received: $msg")
  }
}

// Props without arguments
val props1 = Props[MyActor]()

// Props with constructor arguments
class ParameterizedActor(name: String, value: Int) extends Actor {
  def receive = {
    case msg => println(s"$name($value): $msg")
  }
}

val props2 = Props(new ParameterizedActor("test", 42))

// Props with configuration
val configuredProps = Props[MyActor]()
  .withDispatcher("my-dispatcher")
  .withMailbox("bounded-mailbox")

/**
 * Factory for creating actor references
 */
trait ActorRefFactory {
  /** Create an actor with auto-generated name */
  def actorOf(props: Props): ActorRef
  
  /** Create an actor with specified name */  
  def actorOf(props: Props, name: String): ActorRef
  
  /** Get an actor selection by string path */
  def actorSelection(path: String): ActorSelection
  
  /** Get an actor selection by ActorPath */
  def actorSelection(path: ActorPath): ActorSelection
}

/**
 * Actor context providing creation capabilities within actors
 */
trait ActorContext extends ActorRefFactory {
  /** Reference to this actor */
  def self: ActorRef
  
  /** The actor system this actor belongs to */
  def system: ActorSystem
  
  /** The parent actor reference */
  def parent: ActorRef
  
  /** Iterator over child actor references */
  def children: immutable.Iterable[ActorRef]
  
  /** Lookup child actor by name */
  def child(name: String): Option[ActorRef]
  
  /** Watch another actor for termination */
  def watch(subject: ActorRef): ActorRef
  
  /** Stop watching an actor */
  def unwatch(subject: ActorRef): ActorRef
  
  /** Stop this actor */
  def stop(actor: ActorRef): Unit
  
  /** Become a new behavior */
  def become(behavior: Actor.Receive, discardOld: Boolean = true): Unit
  
  /** Revert to previous behavior */
  def unbecome(): Unit
}

import akka.actor.{ActorSystem, Props}

val system = ActorSystem("MySystem")

// Create actor with generated name
val actor1 = system.actorOf(Props[MyActor]())

// Create actor with specific name  
val actor2 = system.actorOf(Props[MyActor](), "myActor")

// Within an actor, create child actors
class ParentActor extends Actor {
  val child = context.actorOf(Props[MyActor](), "child")
  
  def receive = {
    case "create" => 
      val newChild = context.actorOf(Props[MyActor]())
      context.watch(newChild)
    case Terminated(ref) =>
      println(s"Child $ref terminated")
  }
}

/**
 * Immutable reference to an actor
 */
abstract class ActorRef extends java.lang.Comparable[ActorRef] with Serializable {
  /** Send a message to this actor */
  def tell(msg: Any, sender: ActorRef): Unit
  
  /** Send a message (Scala operator) */
  def !(message: Any)(implicit sender: ActorRef = Actor.noSender): Unit
  
  /** The path of this actor */
  def path: ActorPath
  
  /** Forward a message with original sender */
  def forward(message: Any)(implicit context: ActorContext): Unit
}

object ActorRef {
  /** Special actor reference representing no sender */
  val noSender: ActorRef = Actor.noSender
}

/**
 * Actor path representing the location of an actor
 */
trait ActorPath {
  /** The address of the actor system */
  def address: Address
  
  /** The name of this actor */
  def name: String
  
  /** The parent path */
  def parent: ActorPath
  
  /** The root path */
  def root: RootActorPath
  
  /** Child path with given name */
  def /(child: String): ActorPath
  
  /** Create descendant path */
  def descendant(names: Iterable[String]): ActorPath
  
  /** Convert to string representation */
  override def toString: String
}

/**
 * Address of an actor system
 */
final case class Address(
  protocol: String,
  system: String, 
  host: Option[String] = None,
  port: Option[Int] = None
) {
  /** Whether this address has global scope (network accessible) */
  def hasGlobalScope: Boolean = host.isDefined
  
  /** Whether this address has local scope */
  def hasLocalScope: Boolean = !hasGlobalScope
  
  /** String representation */
  override def toString = {
    val sb = new StringBuilder(protocol)
    sb.append("://").append(system) 
    if (host.isDefined) sb.append('@').append(host.get)
    if (port.isDefined) sb.append(':').append(port.get)
    sb.toString
  }
}

/**
 * Path-based actor lookup supporting wildcards and patterns
 */
abstract class ActorSelection {
  /** Send a message to all matching actors */
  def tell(msg: Any, sender: ActorRef): Unit
  
  /** Send a message (Scala operator) */
  def !(msg: Any)(implicit sender: ActorRef): Unit
  
  /** Forward a message to all matching actors */
  def forward(msg: Any)(implicit context: ActorContext): Unit
  
  /** Resolve to concrete actor references */
  def resolveOne(timeout: FiniteDuration): Future[ActorRef]
  
  /** Get the string path of this selection */
  def pathString: String
  
  /** Create child selection */
  def /(name: String): ActorSelection
}

object ActorSelection {
  /** Create selection from actor reference factory and path */
  def apply(anchorRef: ActorRef, path: String): ActorSelection
  def apply(anchorRef: ActorRef, path: immutable.Iterable[String]): ActorSelection
}

import akka.actor.{ActorSystem, Props}
import scala.concurrent.duration._

val system = ActorSystem("MySystem")

// Actor selection by path
val selection1 = system.actorSelection("/user/myActor")
val selection2 = system.actorSelection("akka://MySystem/user/myActor")

// Wildcard selection
val allChildren = system.actorSelection("/user/*")
val grandchildren = system.actorSelection("/user/*/child")

// Resolve selection to concrete reference
import system.dispatcher
val futureRef = selection1.resolveOne(5.seconds)
futureRef.foreach { ref =>
  println(s"Resolved actor: ${ref.path}")
}

// Within an actor
class MyActor extends Actor {
  def receive = {
    case "find" =>
      val sibling = context.actorSelection("../sibling")
      sibling ! "hello"
  }
}

/**
 * Actor lifecycle hooks - override in actor implementations
 */
trait Actor {
  /** Called before the actor starts processing messages */
  def preStart(): Unit = ()
  
  /** Called after the actor stops processing messages */
  def postStop(): Unit = ()
  
  /** Called before the actor restarts due to an exception */
  def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    context.children.foreach(context.stop)
    postStop()
  }
  
  /** Called after the actor restarts */
  def postRestart(reason: Throwable): Unit = preStart()
  
  /** Supervisor strategy for child actors */
  def supervisorStrategy: SupervisorStrategy = SupervisorStrategy.defaultStrategy
}

import akka.actor.{Actor, ActorLogging}

class LifecycleActor extends Actor with ActorLogging {
  override def preStart(): Unit = {
    log.info("Actor starting up")
    // Initialize resources
  }
  
  override def postStop(): Unit = {
    log.info("Actor shutting down")  
    // Cleanup resources
  }
  
  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    log.error(reason, s"Actor restarting due to error, last message: $message")
    super.preRestart(reason, message)
  }
  
  override def postRestart(reason: Throwable): Unit = {
    log.info("Actor restarted successfully")
    super.postRestart(reason)
  }
  
  def receive = {
    case "crash" => throw new RuntimeException("Intentional crash")
    case msg => log.info(s"Received: $msg")
  }
}

/**
 * Lifecycle control messages
 */
case object PoisonPill extends AutoReceivedMessage with PossiblyHarmful
case object Kill extends AutoReceivedMessage with PossiblyHarmful

/**
 * Termination notification message
 */
final case class Terminated(actor: ActorRef)(
  val existenceConfirmed: Boolean,
  val addressTerminated: Boolean  
) extends AutoReceivedMessage with PossiblyHarmful

/**
 * Graceful stop pattern from akka.pattern
 */
object GracefulStopSupport {
  def gracefulStop(
    actorRef: ActorRef,
    timeout: FiniteDuration,
    stopMessage: Any = PoisonPill
  )(implicit system: ActorSystem): Future[Boolean]
}

final case class GracefulStopTimeoutException(message: String) extends AkkaException(message)

import akka.actor.{Actor, ActorRef, PoisonPill, Kill, Terminated}
import akka.pattern.gracefulStop
import scala.concurrent.duration._

class WatchingActor extends Actor {
  var childRef: ActorRef = _
  
  override def preStart(): Unit = {
    childRef = context.actorOf(Props[MyActor](), "child")
    context.watch(childRef) // Watch for termination
  }
  
  def receive = {
    case "stop-child" => 
      childRef ! PoisonPill // Graceful termination
      
    case "kill-child" =>
      childRef ! Kill // Forceful termination with exception
      
    case Terminated(ref) =>
      println(s"Child $ref has terminated")
      
    case "stop-self" =>
      context.stop(self)
  }
}

// Graceful stop from outside
import system.dispatcher
val stopFuture = gracefulStop(actorRef, 5.seconds)
stopFuture.onSuccess {
  case true => println("Actor stopped gracefully")
  case false => println("Actor did not stop within timeout")
}

/**
 * Deployment configuration for actors
 */
final case class Deploy(
  path: String = "",
  config: Config = ConfigFactory.empty(),
  routerConfig: RouterConfig = NoRouter,
  scope: Scope = NoScopeGiven,
  dispatcher: String = Deploy.NoDispatcherGiven,
  mailbox: String = Deploy.NoMailboxGiven
) {
  def withFallback(other: Deploy): Deploy
  def withDispatcher(dispatcher: String): Deploy
  def withMailbox(mailbox: String): Deploy
  def withRouterConfig(routerConfig: RouterConfig): Deploy
}

object Deploy {
  val NoDispatcherGiven = ""
  val NoMailboxGiven = ""
  val NoScopeGiven: Scope = NoScopeGiven
}

/**
 * Deployment scope for actor placement
 */
abstract class Scope
case object LocalScope extends Scope
case object NoScopeGiven extends Scope

import akka.actor.{Deploy, Props}
import akka.routing.RoundRobinPool

// Configure deployment in Props
val deployedProps = Props[MyActor]().withDeploy(
  Deploy()
    .withDispatcher("my-dispatcher")
    .withMailbox("bounded-mailbox") 
    .withRouterConfig(RoundRobinPool(5))
)

val routerActor = system.actorOf(deployedProps, "router")