use crate::context::{ActorControlHandle, ActorControlMessage, Context};
use crate::error::TheatreError;
use crate::mailbox::{Addr, Mailbox};
use async_trait::async_trait;
use futures::channel::oneshot;
use futures::SinkExt;
use tokio::task::{JoinError, JoinHandle};

/// Actor is one of the core traits of theatre.
///
/// To create an actor just implement the trait for your struct
///
/// Example:
/// ```rust
/// use theatre::prelude::*;
///
/// struct MyActor;
///
/// impl Actor for MyActor {}
/// ```
///
/// Actor has methods that can be optionally overriden to track actor lifecycle.
///
/// The actor's event loop is shutdown when all addresses to the actor are dropped.
#[async_trait]
pub trait Actor: Send {
    async fn starting(&mut self, _ctx: &mut Context<Self>)
    where
        Self: Sized,
    {
    }

    async fn stopping(&mut self, _ctx: &mut Context<Self>)
    where
        Self: Sized,
    {
    }

    async fn stopped(&mut self, _ctx: &mut Context<Self>)
    where
        Self: Sized,
    {
    }
}

/// ActorHandle is a method of `await`ing the actor's task. Dropping the [ActorHandle] will not
/// shutdown the actor.
///
/// #[ActorHandle](ActorHandle)
#[derive(Debug)]
pub struct ActorHandle {
    ctrl_handle: ActorControlHandle,
    task: JoinHandle<()>,
}

impl ActorHandle {
    pub(crate) fn new(ctrl_handle: ActorControlHandle, task: JoinHandle<()>) -> Self {
        Self { ctrl_handle, task }
    }

    /// Shutdown the actor
    pub fn shutdown(&mut self) -> Result<(), TheatreError> {
        self.ctrl_handle
            .try_send(ActorControlMessage::Shutdown)
            .map_err(|_| TheatreError::SendError)
    }

    /// Calling `join()` will `await` the actor until it has shutdown.
    pub async fn join(self) -> Result<(), JoinError> {
        self.task.await
    }

    /// Send a heartbeat message to the actor. The actor will response as soon as it can.
    /// This method returns true if the actor is still running or false if the actor has shutdown.
    pub async fn heartbeat(&mut self) -> bool {
        let (tx, rx) = oneshot::channel();
        if self
            .ctrl_handle
            .send(ActorControlMessage::Heartbeat(tx))
            .await
            .is_err()
        {
            return false;
        }

        let reply = rx.await;

        reply.is_ok()
    }
}

/// Start an actor in a dedicated task and return its address and handle.
pub async fn start<A>(actor: A) -> (Addr<A>, ActorHandle)
where
    A: 'static + Actor + Sized,
{
    let (addr_tx, addr_rx) = oneshot::channel();
    let task = tokio::spawn(async move {
        let mailbox = Mailbox::new(16);

        let addr = mailbox.new_addr();
        let (mut actor_ctx, handle) = Context::new(mailbox);

        addr_tx.send((addr, handle)).ok().unwrap();

        // Ignore any panics
        let _ = actor_ctx.run(actor).await;
    });

    let (addr, ctrl_handle) = addr_rx.await.unwrap();
    (addr, ActorHandle { ctrl_handle, task })
}

/// Extensions for Actors. This is blanket impl'd on any type that implements `Actor`.
#[async_trait]
pub trait ActorExt {
    /// Start a new actor. Return its address and discard the handle
    async fn spawn(self) -> Addr<Self>
    where
        Self: 'static + Actor + Sized;

    /// Start a new actor. Return its address and handle.
    async fn start(self) -> (Addr<Self>, ActorHandle)
    where
        Self: 'static + Actor + Sized;
}

#[async_trait]
impl<A> ActorExt for A
where
    A: Actor,
{
    async fn spawn(self) -> Addr<Self>
    where
        A: 'static + Actor + Sized,
    {
        start(self).await.0
    }

    async fn start(self) -> (Addr<Self>, ActorHandle)
    where
        Self: 'static + Actor + Sized,
    {
        start(self).await
    }
}

#[cfg(test)]
mod test {
    use super::Actor;
    use futures::channel::oneshot;
    use futures::channel::oneshot::Sender;

    struct A;
    impl Actor for A {}

    #[tokio::test]
    async fn start_actor_and_shutdown() {
        let mut handle = crate::actor::start(A).await.1;
        handle.shutdown().unwrap()
    }

    #[tokio::test]
    async fn start_actor_and_join() {
        let mut handle = crate::actor::start(A).await.1;
        handle.shutdown().unwrap();

        handle.join().await.unwrap()
    }

    #[tokio::test]
    async fn start_with_actor_ext() {
        use super::ActorExt;

        let mut handle = A.start().await.1;
        handle.shutdown().unwrap();
    }

    #[tokio::test]
    async fn spawn_with_actor_ext() {
        use super::ActorExt;

        let _ = A.spawn();
    }

    #[tokio::test]
    async fn actor_starting() {
        use crate::prelude::*;

        let (start, started) = oneshot::channel();

        struct ReplyOnStarting {
            start: Option<Sender<bool>>,
        }

        #[async_trait]
        impl Actor for ReplyOnStarting {
            async fn starting(&mut self, _: &mut Context<Self>) {
                self.start.take().unwrap().send(true).unwrap();
            }
        }

        let _ = ReplyOnStarting { start: Some(start) }.start().await;

        let started = started.await.unwrap();

        assert_eq!(started, true);
    }

    #[tokio::test]
    async fn actor_stopping() {
        use crate::prelude::*;

        let (stop, stopping) = oneshot::channel();

        struct ReplyOnStopping {
            stopping: Option<Sender<bool>>,
        }

        #[async_trait]
        impl Actor for ReplyOnStopping {
            async fn stopping(&mut self, _: &mut Context<Self>) {
                self.stopping.take().unwrap().send(true).unwrap();
            }
        }

        let addr = ReplyOnStopping {
            stopping: Some(stop),
        }
        .start()
        .await;
        drop(addr);

        let stopping = stopping.await.unwrap();

        assert_eq!(stopping, true);
    }

    #[tokio::test]
    async fn actor_stopped() {
        use crate::prelude::*;

        let (stop, stopped) = oneshot::channel();

        struct ReplyOnStopped {
            stopped: Option<Sender<bool>>,
        }

        #[async_trait]
        impl Actor for ReplyOnStopped {
            async fn stopped(&mut self, _: &mut Context<Self>) {
                self.stopped.take().unwrap().send(true).unwrap();
            }
        }

        let addr = ReplyOnStopped {
            stopped: Some(stop),
        }
        .start()
        .await;
        drop(addr);

        let stopped = stopped.await.unwrap();
        assert_eq!(stopped, true);
    }
}