near-async 0.34.3

use std::any::type_name;
use std::fmt::Debug;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use crate::futures::DelayedActionRunner;
use crate::messaging::{Actor, AsyncSendError, CanSend, CanSendAsync, HandlerWithContext};
use crate::time::Duration;

use super::PendingEventsSender;
use super::data::{TestLoopData, TestLoopDataHandle};
use futures::FutureExt;
use futures::channel::oneshot;
use futures::future::BoxFuture;

/// TestLoopSender implements the CanSend methods for an actor that can Handle them.
///
/// ```rust, ignore
/// let actor = TestActor::new();
/// let adapter = LateBoundSender::new();
///
/// let sender: TestLoopSender<TestActor> = data.register_actor(actor, Some(adapter));
///
/// // We can now send messages to the actor using the sender and adapter.
/// sender.send(TestMessage {});
/// adapter.send(TestMessage {});
/// ```
///
/// For the purposes of testloop, we keep a copy of the delay sender that is used to schedule
/// callbacks on the testloop to execute either the actor.handle() function or the
/// DelayedActionRunner.run_later_boxed() function.
pub struct TestLoopSender<A>
where
    A: 'static,
{
    actor_handle: TestLoopDataHandle<A>,
    pending_events_sender: PendingEventsSender,
    shutting_down: Arc<AtomicBool>,
    sender_delay: Duration,
}

impl<A> Clone for TestLoopSender<A> {
    fn clone(&self) -> Self {
        Self {
            actor_handle: self.actor_handle.clone(),
            pending_events_sender: self.pending_events_sender.clone(),
            shutting_down: self.shutting_down.clone(),
            sender_delay: self.sender_delay,
        }
    }
}

/// `DelayedActionRunner` that schedules the action to be run later by the TestLoop event loop.
impl<A> DelayedActionRunner<A> for TestLoopSender<A>
where
    A: 'static,
{
    fn run_later_boxed(
        &mut self,
        name: &str,
        dur: Duration,
        f: Box<dyn FnOnce(&mut A, &mut dyn DelayedActionRunner<A>) + Send + 'static>,
    ) {
        if self.shutting_down.load(Ordering::Relaxed) {
            return;
        }

        let mut this = self.clone();
        let callback = move |data: &mut TestLoopData| {
            let actor = data.get_mut(&this.actor_handle);
            f(actor, &mut this);
        };
        self.pending_events_sender.send_with_delay(
            format!("DelayedAction {}({:?})", pretty_type_name::<A>(), name),
            Box::new(callback),
            dur,
        );
    }
}

impl<M, A> CanSend<M> for TestLoopSender<A>
where
    M: Debug + Send + 'static,
    A: Actor + HandlerWithContext<M> + 'static,
{
    fn send(&self, msg: M) {
        let mut this = self.clone();
        let description = format!("{}({:?})", pretty_type_name::<A>(), &msg);
        let callback = move |data: &mut TestLoopData| {
            let actor = data.get_mut(&this.actor_handle);
            actor.handle(msg, &mut this);
        };
        self.pending_events_sender.send_with_delay(
            description,
            Box::new(callback),
            self.sender_delay,
        );
    }
}

impl<M, R, A> CanSendAsync<M, R> for TestLoopSender<A>
where
    M: Debug + Send + 'static,
    A: Actor + HandlerWithContext<M, R> + 'static,
    R: Send + 'static,
{
    fn send_async(&self, msg: M) -> BoxFuture<'static, Result<R, AsyncSendError>> {
        let mut this = self.clone();
        let description = format!("{}({:?})", pretty_type_name::<A>(), &msg);
        let (sender, receiver) = oneshot::channel::<R>();
        let callback = move |data: &mut TestLoopData| {
            let actor = data.get_mut(&this.actor_handle);
            let result = actor.handle(msg, &mut this);
            sender.send(result).ok();
        };
        self.pending_events_sender.send_with_delay(
            description,
            Box::new(callback),
            self.sender_delay,
        );
        async move { Ok(receiver.await.unwrap()) }.boxed()
    }
}

impl<A> TestLoopSender<A>
where
    A: Actor + 'static,
{
    pub(crate) fn new(
        actor_handle: TestLoopDataHandle<A>,
        pending_events_sender: PendingEventsSender,
        shutting_down: Arc<AtomicBool>,
    ) -> Self {
        Self { actor_handle, pending_events_sender, shutting_down, sender_delay: Duration::ZERO }
    }

    /// Returns a new TestLoopSender which sends messages with the given delay.
    pub fn with_delay(self, delay: Duration) -> Self {
        Self { sender_delay: delay, ..self }
    }

    pub fn actor_handle(&self) -> TestLoopDataHandle<A> {
        self.actor_handle.clone()
    }
}

// Quick and dirty way of getting the type name without the module path.
// Does not work for more complex types like std::sync::Arc<std::sync::atomic::AtomicBool<...>>
// example near_chunks::shards_manager_actor::ShardsManagerActor -> ShardsManagerActor
fn pretty_type_name<T>() -> &'static str {
    type_name::<T>().split("::").last().unwrap()
}