use crate::actor::Actor;
use crate::mailbox::Addr;
use crate::mailbox::Mailbox;
use crate::mailbox::MailboxError;
use futures::channel::mpsc;
use futures::channel::oneshot::Sender;
use futures::future::pending;
use futures::select;
use futures::stream::{Fuse, StreamExt};
use futures::FutureExt;
use std::any::Any;
use std::panic::AssertUnwindSafe;

#[derive(Copy, Clone, Debug, PartialOrd, PartialEq, Ord, Eq)]
enum ActorState {
    Bootstrap,
    Running,
    Stopping,
    Stopped,
}

#[derive(Debug)]
pub(crate) enum ActorControlMessage {
    Shutdown,
    Heartbeat(Sender<()>),
}

pub(crate) type ActorControlHandle = mpsc::Sender<ActorControlMessage>;
/// Context is the execution context of the actor. It allows for the actor to dynamically retrieve
/// the address of the actor's `Mailbox`.
///
/// It also provides a method to allow the actor to stop itself.
#[derive(Debug)]
pub struct Context<A>
where
    A: 'static + Actor,
{
    mailbox: Mailbox<A>,
    state: ActorState,
    control_channel: Fuse<mpsc::Receiver<ActorControlMessage>>,
}

impl<A> Context<A>
where
    A: Actor,
{
    pub(crate) fn new(mailbox: Mailbox<A>) -> (Self, ActorControlHandle) {
        let (tx, rx) = mpsc::channel(16);
        (
            Self {
                mailbox,
                state: ActorState::Bootstrap,
                control_channel: rx.fuse(),
            },
            tx,
        )
    }

    /// Run the actor event loop. Any panics by handlers are caught and returned as Result::Err.
    pub(crate) async fn run(&mut self, mut actor: A) -> Result<(), Box<dyn Any + Send>> {
        actor.starting(self).await;
        self.state = ActorState::Running;

        while self.state == ActorState::Running {
            select! {
                maybe_envelope = self.mailbox.recv().fuse() => match maybe_envelope {
                    Ok(mut envelope) => AssertUnwindSafe(envelope.process(&mut actor, self)).catch_unwind().await?,
                    // If the mailbox is closed, stop the event loop
                    Err(mailbox_err) => match mailbox_err {
                        MailboxError::Closed => {
                            break
                        }
                    }
                },
                // If the control handle has been dropped, don't immediately continue the loop
                ctrl_msg = self.control_channel.next().then(|x| async move { match x { Some(s) => s, None => pending::<_>().await }}) => {
                    self.control(ctrl_msg)
                }
            }
        }

        self.state = ActorState::Stopping;

        // Notify the actor that it is  stopping
        actor.stopping(self).await;

        // Close the mailbox of the actor if it isn't already closed.
        self.mailbox.close();

        // Drain any messages remaining in the mailbox
        while let Ok(mut envelope) = self.mailbox.recv().await {
            envelope.process(&mut actor, self).await;
        }

        self.state = ActorState::Stopped;
        actor.stopped(self).await;
        Ok(())
    }

    /// Get the address of the mailbox
    pub async fn address(&mut self) -> Addr<A> {
        self.mailbox.new_addr()
    }

    pub fn shutdown(&mut self) {
        self.state = ActorState::Stopping;
    }

    fn control(&mut self, msg: ActorControlMessage) {
        match msg {
            ActorControlMessage::Shutdown => self.shutdown(),
            ActorControlMessage::Heartbeat(tx) => {
                let _ = tx.send(());
            }
        }
    }
}
