/*** Construct an [[akka.actor.ActorSelection]] from the given path, which is* parsed for wildcards (these are replaced by regular expressions* internally). No attempt is made to verify the existence of any part of* the supplied path, it is recommended to send a message and gather the* replies in order to resolve the matching set of actors.*/def actorSelection(path: String): ActorSelection = path match {case RelativeActorPath(elems) ⇒if (elems.isEmpty) ActorSelection(provider.deadLetters, "")else if (elems.head.isEmpty) ActorSelection(provider.rootGuardian, elems.tail)else ActorSelection(lookupRoot, elems)case ActorPathExtractor(address, elems) ⇒ActorSelection(provider.rootGuardianAt(address), elems)case _ ⇒ActorSelection(provider.deadLetters, "")}

/*** Extractor for so-called “relative actor paths” as in “relative URI”, not in* “relative to some actor”. Examples:**  * "grand/child"*  * "/user/hello/world"*/
object RelativeActorPath extends PathUtils {def unapply(addr: String): Option[immutable.Seq[String]] = {try {val uri = new URI(addr)if (uri.isAbsolute) Noneelse Some(split(uri.getRawPath, uri.getRawFragment))} catch {case _: URISyntaxException ⇒ None}}
}

/*** Given an ActorPath it returns the Address and the path elements if the path is well-formed*/
object ActorPathExtractor extends PathUtils {def unapply(addr: String): Option[(Address, immutable.Iterable[String])] =try {val uri = new URI(addr)uri.getRawPath match {case null ⇒ Nonecase path ⇒ AddressFromURIString.unapply(uri).map((_, split(path, uri.getRawFragment).drop(1)))}} catch {case _: URISyntaxException ⇒ None}
}

  override def rootGuardianAt(address: Address): ActorRef =if (address == rootPath.address) rootGuardianelse deadLetters

def rootGuardianAt(address: Address): ActorRef = {if (hasAddress(address)) rootGuardianelse try {new RemoteActorRef(transport, transport.localAddressForRemote(address),RootActorPath(address), Nobody, props = None, deploy = None)} catch {case NonFatal(e) ⇒log.error(e, "No root guardian at [{}]", address)new EmptyLocalActorRef(this, RootActorPath(address), eventStream)}}

/*** Construct an ActorSelection from the given string representing a path* relative to the given target. This operation has to create all the* matching magic, so it is preferable to cache its result if the* intention is to send messages frequently.*/def apply(anchorRef: ActorRef, elements: Iterable[String]): ActorSelection = {val compiled: immutable.IndexedSeq[SelectionPathElement] = elements.collect({case x if !x.isEmpty ⇒if ((x.indexOf('?') != -1) || (x.indexOf('*') != -1)) SelectChildPattern(x)else if (x == "..") SelectParentelse SelectChildName(x)})(scala.collection.breakOut)new ActorSelection with ScalaActorSelection {override val anchor = anchorRefoverride val path = compiled}}

  def tell(msg: Any, sender: ActorRef): Unit =ActorSelection.deliverSelection(anchor.asInstanceOf[InternalActorRef], sender,ActorSelectionMessage(msg, path, wildcardFanOut = false))

private[akka] class RemoteActorRef private[akka] (remote:                RemoteTransport,val localAddressToUse: Address,val path:              ActorPath,val getParent:         InternalActorRef,props:                 Option[Props],deploy:                Option[Deploy])extends InternalActorRef with RemoteRef

override def !(message: Any)(implicit sender: ActorRef = Actor.noSender): Unit = {if (message == null) throw InvalidMessageException("Message is null")try remote.send(message, OptionVal(sender), this) catch handleException(message, sender)}

override def send(message: Any, senderOption: OptionVal[ActorRef], recipient: RemoteActorRef): Unit = endpointManager match {case Some(manager) ⇒ manager.tell(Send(message, senderOption, recipient), sender = senderOption getOrElse Actor.noSender)case None          ⇒ throw new RemoteTransportExceptionNoStackTrace("Attempted to send remote message but Remoting is not running.", null)}

case s @ Send(message, senderOption, recipientRef, _) ⇒val recipientAddress = recipientRef.path.addressdef createAndRegisterWritingEndpoint(): ActorRef = {endpoints.registerWritableEndpoint(recipientAddress,uid = None,createEndpoint(recipientAddress,recipientRef.localAddressToUse,transportMapping(recipientRef.localAddressToUse),settings,handleOption = None,writing = true))}endpoints.writableEndpointWithPolicyFor(recipientAddress) match {case Some(Pass(endpoint, _)) ⇒endpoint ! scase Some(Gated(timeOfRelease)) ⇒if (timeOfRelease.isOverdue()) createAndRegisterWritingEndpoint() ! selse extendedSystem.deadLetters ! scase Some(Quarantined(uid, _)) ⇒// timeOfRelease is only used for garbage collection reasons, therefore it is ignored here. We still have// the Quarantined tombstone and we know what UID we don't want to accept, so use it.createAndRegisterWritingEndpoint() ! scase None ⇒createAndRegisterWritingEndpoint() ! s}

def registerWritableEndpoint(address: Address, uid: Option[Int], endpoint: ActorRef): ActorRef =addressToWritable.get(address) match {case Some(Pass(e, _)) ⇒throw new IllegalArgumentException(s"Attempting to overwrite existing endpoint [$e] with [$endpoint]")case _ ⇒// note that this overwrites Quarantine marker,// but that is ok since we keep the quarantined uid in addressToRefuseUidaddressToWritable += address → Pass(endpoint, uid)writableToAddress += endpoint → addressendpoint}

private def createEndpoint(remoteAddress:    Address,localAddress:     Address,transport:        AkkaProtocolTransport,endpointSettings: RemoteSettings,handleOption:     Option[AkkaProtocolHandle],writing:          Boolean): ActorRef = {require(transportMapping contains localAddress, "Transport mapping is not defined for the address")// refuseUid is ignored for read-only endpoints since the UID of the remote system is already known and has passed// quarantine checksval refuseUid = endpoints.refuseUid(remoteAddress)if (writing) context.watch(context.actorOf(RARP(extendedSystem).configureDispatcher(ReliableDeliverySupervisor.props(handleOption,localAddress,remoteAddress,refuseUid,transport,endpointSettings,AkkaPduProtobufCodec,receiveBuffers)).withDeploy(Deploy.local),"reliableEndpointWriter-" + AddressUrlEncoder(remoteAddress) + "-" + endpointId.next()))else context.watch(context.actorOf(RARP(extendedSystem).configureDispatcher(EndpointWriter.props(handleOption,localAddress,remoteAddress,refuseUid,transport,endpointSettings,AkkaPduProtobufCodec,receiveBuffers,reliableDeliverySupervisor = None)).withDeploy(Deploy.local),"endpointWriter-" + AddressUrlEncoder(remoteAddress) + "-" + endpointId.next()))}

  private def handleSend(send: Send): Unit =if (send.message.isInstanceOf[SystemMessage]) {val sequencedSend = send.copy(seqOpt = Some(nextSeq()))tryBuffer(sequencedSend)// If we have not confirmed the remote UID we cannot transfer the system message at this point just buffer it.// GotUid will kick resendAll() causing the messages to be properly written.// Flow control by not sending more when we already have many outstanding.if (uidConfirmed && resendBuffer.nonAcked.size <= settings.SysResendLimit)writer ! sequencedSend} else writer ! send

var writer: ActorRef = createWriter()

private def createWriter(): ActorRef = {context.watch(context.actorOf(RARP(context.system).configureDispatcher(EndpointWriter.props(handleOrActive = currentHandle,localAddress = localAddress,remoteAddress = remoteAddress,refuseUid,transport = transport,settings = settings,AkkaPduProtobufCodec,receiveBuffers = receiveBuffers,reliableDeliverySupervisor = Some(self))).withDeploy(Deploy.local), "endpointWriter"))}

def receive = if (handle.isEmpty) initializing else writing

val writing: Receive = {case s: Send ⇒if (!writeSend(s)) {enqueueInBuffer(s)scheduleBackoffTimer()context.become(buffering)}// We are in Writing state, so buffer is empty, safe to stop herecase FlushAndStop ⇒flushAndStop()case AckIdleCheckTimer if ackDeadline.isOverdue() ⇒trySendPureAck()}

var handle: Option[AkkaProtocolHandle] = handleOrActive

private[remote] class AkkaProtocolHandle(_localAddress:          Address,_remoteAddress:         Address,val readHandlerPromise: Promise[HandleEventListener],_wrappedHandle:         AssociationHandle,val handshakeInfo:      HandshakeInfo,private val stateActor: ActorRef,private val codec:      AkkaPduCodec)extends AbstractTransportAdapterHandle(_localAddress, _remoteAddress, _wrappedHandle, AkkaScheme) {override def write(payload: ByteString): Boolean = wrappedHandle.write(codec.constructPayload(payload))override def disassociate(): Unit = disassociate(Unknown)def disassociate(info: DisassociateInfo): Unit = stateActor ! DisassociateUnderlying(info)
}

/*** A message all Actors will understand, that when processed will reply with* [[akka.actor.ActorIdentity]] containing the `ActorRef`. The `messageId`* is returned in the `ActorIdentity` message as `correlationId`.*/
@SerialVersionUID(1L)
final case class Identify(messageId: Any) extends AutoReceivedMessage with NotInfluenceReceiveTimeout

//Memory consistency is handled by the Mailbox (reading mailbox status then processing messages, then writing mailbox statusfinal def invoke(messageHandle: Envelope): Unit = {val influenceReceiveTimeout = !messageHandle.message.isInstanceOf[NotInfluenceReceiveTimeout]try {currentMessage = messageHandleif (influenceReceiveTimeout)cancelReceiveTimeout()messageHandle.message match {case msg: AutoReceivedMessage ⇒ autoReceiveMessage(messageHandle)case msg                      ⇒ receiveMessage(msg)}currentMessage = null // reset current message after successful invocation} catch handleNonFatalOrInterruptedException { e ⇒handleInvokeFailure(Nil, e)} finally {if (influenceReceiveTimeout)checkReceiveTimeout // Reschedule receive timeout}}def autoReceiveMessage(msg: Envelope): Unit = {if (system.settings.DebugAutoReceive)publish(Debug(self.path.toString, clazz(actor), "received AutoReceiveMessage " + msg))msg.message match {case t: Terminated              ⇒ receivedTerminated(t)case AddressTerminated(address) ⇒ addressTerminated(address)case Kill                       ⇒ throw ActorKilledException("Kill")case PoisonPill                 ⇒ self.stop()case sel: ActorSelectionMessage ⇒ receiveSelection(sel)case Identify(messageId)        ⇒ sender() ! ActorIdentity(messageId, Some(self))}}