/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * license agreements; and to You under the Apache License, version 2.0:
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * This file is part of the Apache Pekko project, which was derived from Akka.
 */

/*
 * Copyright (C) 2014-2022 Lightbend Inc. <https://www.lightbend.com>
 */

package org.apache.pekko.actor.typed

import scala.annotation.unchecked.uncheckedVariance

import org.apache.pekko
import pekko.{ actor => classic }
import pekko.actor.typed.internal.InternalRecipientRef
import pekko.annotation.DoNotInherit

/**
 * An ActorRef is the identity or address of an Actor instance. It is valid
 * only during the Actor’s lifetime and allows messages to be sent to that
 * Actor instance. Sending a message to an Actor that has terminated before
 * receiving the message will lead to that message being discarded; such
 * messages are delivered to the [[DeadLetter]] channel of the
 * [[pekko.event.EventStream]] on a best effort basis
 * (i.e. this delivery is not reliable).
 *
 * Not for user extension
 */
@DoNotInherit
trait ActorRef[-T] extends RecipientRef[T] with java.lang.Comparable[ActorRef[_]] with java.io.Serializable {
  this: InternalRecipientRef[T] =>

  /**
   * Send a message to the Actor referenced by this ActorRef using *at-most-once*
   * messaging semantics.
   */
  def tell(msg: T): Unit

  /**
   * Narrow the type of this `ActorRef`, which is always a safe operation.
   */
  def narrow[U <: T]: ActorRef[U]

  /**
   * Unsafe utility method for widening the type accepted by this ActorRef;
   * provided to avoid having to use `asInstanceOf` on the full reference type,
   * which would unfortunately also work on non-ActorRefs. Use it with caution,it may cause a [[ClassCastException]] when you send a message
   * to the widened [[ActorRef[U]]].
   */
  def unsafeUpcast[U >: T @uncheckedVariance]: ActorRef[U]

  /**
   * The hierarchical path name of the referenced Actor. The lifecycle of the
   * ActorRef is fully contained within the lifecycle of the [[pekko.actor.ActorPath]]
   * and more than one Actor instance can exist with the same path at different
   * points in time, but not concurrently.
   */
  def path: classic.ActorPath

  @throws(classOf[java.io.ObjectStreamException])
  private def writeReplace(): AnyRef = SerializedActorRef[T](this)
}

object ActorRef {

  implicit final class ActorRefOps[-T](val ref: ActorRef[T]) extends AnyVal {

    /**
     * Send a message to the Actor referenced by this ActorRef using *at-most-once*
     * messaging semantics.
     */
    def !(msg: T): Unit = ref.tell(msg)
  }
}

/**
 * INTERNAL API
 */
private[pekko] object SerializedActorRef {
  def apply[T](actorRef: ActorRef[T]): SerializedActorRef[T] = {
    new SerializedActorRef(actorRef)
  }

  def toAddress[T](actorRef: ActorRef[T]) = {
    import pekko.actor.typed.scaladsl.adapter._
    import pekko.serialization.JavaSerializer.currentSystem
    val resolver = ActorRefResolver(currentSystem.value.toTyped)
    resolver.toSerializationFormat(actorRef)
  }
}

/**
 * Memento pattern for serializing ActorRefs transparently
 * INTERNAL API
 */
@SerialVersionUID(1L)
private[pekko] final case class SerializedActorRef[T] private (address: String) {
  import pekko.actor.typed.scaladsl.adapter._
  import pekko.serialization.JavaSerializer.currentSystem

  def this(actorRef: ActorRef[T]) =
    this(SerializedActorRef.toAddress(actorRef))

  @throws(classOf[java.io.ObjectStreamException])
  def readResolve(): AnyRef = currentSystem.value match {
    case null =>
      throw new IllegalStateException(
        "Trying to deserialize a serialized typed ActorRef without an ActorSystem in scope." +
        " Use 'org.apache.pekko.serialization.Serialization.currentSystem.withValue(system) { ... }'")
    case someSystem =>
      val resolver = ActorRefResolver(someSystem.toTyped)
      resolver.resolveActorRef(address)
  }
}

/**
 * FIXME doc
 * - not serializable
 * - not watchable
 */
trait RecipientRef[-T] { this: InternalRecipientRef[T] =>

  /**
   * Send a message to the destination referenced by this `RecipientRef` using *at-most-once*
   * messaging semantics.
   */
  def tell(msg: T): Unit
}

object RecipientRef {

  implicit final class RecipientRefOps[-T](val ref: RecipientRef[T]) extends AnyVal {

    /**
     * Send a message to the destination referenced by this `RecipientRef` using *at-most-once*
     * messaging semantics.
     */
    def !(msg: T): Unit = ref.tell(msg)
  }
}
