use crate::functional::{SendAsyncFunction, SendFunction};
use crate::futures::DelayedActionRunner;
use futures::FutureExt;
use futures::future::BoxFuture;
use std::fmt::{Debug, Display};
use std::sync::{Arc, OnceLock};
pub trait Actor {
fn start_actor(&mut self, _ctx: &mut dyn DelayedActionRunner<Self>) {}
fn wrap_handler<M, R>(
&mut self,
msg: M,
ctx: &mut dyn DelayedActionRunner<Self>,
f: impl FnOnce(&mut Self, M, &mut dyn DelayedActionRunner<Self>) -> R,
) -> R {
f(self, msg, ctx)
}
fn stop_actor(&mut self) {}
}
pub trait Handler<M, R = ()>
where
M: Send + 'static,
R: Send,
{
fn handle(&mut self, msg: M) -> R;
}
pub trait HandlerWithContext<M, R = ()>
where
M: Send + 'static,
R: Send,
{
fn handle(&mut self, msg: M, ctx: &mut dyn DelayedActionRunner<Self>) -> R;
}
impl<A, M, R> HandlerWithContext<M, R> for A
where
A: Actor + Handler<M, R>,
M: Send + 'static,
R: Send,
{
fn handle(&mut self, msg: M, ctx: &mut dyn DelayedActionRunner<Self>) -> R {
self.wrap_handler(msg, ctx, |this, msg, _| Handler::handle(this, msg))
}
}
pub trait CanSend<M>: Send + Sync + 'static {
fn send(&self, message: M);
}
pub trait CanSendAsync<M, R>: Send + Sync + 'static {
fn send_async(&self, message: M) -> BoxFuture<'static, Result<R, AsyncSendError>>;
}
pub struct Sender<M: 'static> {
sender: Arc<dyn CanSend<M>>,
}
impl<M> Clone for Sender<M> {
fn clone(&self) -> Self {
Self { sender: self.sender.clone() }
}
}
pub trait IntoSender<M> {
fn into_sender(self) -> Sender<M>;
fn as_sender(self: &Arc<Self>) -> Sender<M>;
}
impl<M, T: CanSend<M>> IntoSender<M> for T {
fn into_sender(self) -> Sender<M> {
Sender::from_impl(self)
}
fn as_sender(self: &Arc<Self>) -> Sender<M> {
Sender::from_arc(self.clone())
}
}
impl<M> Sender<M> {
pub fn send(&self, message: M) {
self.sender.send(message)
}
fn from_impl(sender: impl CanSend<M> + 'static) -> Self {
Self { sender: Arc::new(sender) }
}
fn from_arc<T: CanSend<M> + 'static>(arc: Arc<T>) -> Self {
Self { sender: arc }
}
pub fn from_fn(send: impl Fn(M) + Send + Sync + 'static) -> Self {
Self::from_impl(SendFunction::new(send))
}
}
pub struct AsyncSender<M: 'static, R: Send + 'static> {
sender: Arc<dyn CanSendAsync<M, R>>,
}
impl<M, R: Send + 'static> Clone for AsyncSender<M, R> {
fn clone(&self) -> Self {
Self { sender: self.sender.clone() }
}
}
pub trait IntoAsyncSender<M, R: Send> {
fn into_async_sender(self) -> AsyncSender<M, R>;
fn as_async_sender(self: &Arc<Self>) -> AsyncSender<M, R>;
}
impl<M, R: Send + 'static, T: CanSendAsync<M, R>> IntoAsyncSender<M, R> for T {
fn into_async_sender(self) -> AsyncSender<M, R> {
AsyncSender::from_impl(self)
}
fn as_async_sender(self: &Arc<Self>) -> AsyncSender<M, R> {
AsyncSender::from_arc(self.clone())
}
}
impl<M, R: Send + 'static> AsyncSender<M, R> {
pub fn send_async(&self, message: M) -> BoxFuture<'static, Result<R, AsyncSendError>> {
self.sender.send_async(message)
}
fn from_impl(sender: impl CanSendAsync<M, R> + 'static) -> Self {
Self { sender: Arc::new(sender) }
}
fn from_arc<T: CanSendAsync<M, R> + 'static>(arc: Arc<T>) -> Self {
Self { sender: arc }
}
pub fn from_fn(send: impl Fn(M) -> R + Send + Sync + 'static) -> Self {
Self::from_impl(SendAsyncFunction::new(send))
}
}
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum AsyncSendError {
Closed,
Timeout,
Dropped,
}
impl std::error::Error for AsyncSendError {}
impl Display for AsyncSendError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
Debug::fmt(self, f)
}
}
pub struct LateBoundSender<S> {
sender: OnceLock<S>,
}
impl<S> LateBoundSender<S> {
pub fn new() -> Arc<Self> {
Arc::new(Self { sender: OnceLock::new() })
}
pub fn bind(&self, sender: S) {
self.sender.set(sender).map_err(|_| ()).expect("cannot set sender twice");
}
}
impl<M, S: CanSend<M>> CanSend<M> for LateBoundSender<S> {
fn send(&self, message: M) {
self.sender.wait().send(message);
}
}
impl<M, R, S: CanSendAsync<M, R>> CanSendAsync<M, R> for LateBoundSender<S> {
fn send_async(&self, message: M) -> BoxFuture<'static, Result<R, AsyncSendError>> {
self.sender.wait().send_async(message)
}
}
pub struct Noop;
impl<M> CanSend<M> for Noop {
fn send(&self, _message: M) {}
}
impl<M> CanSend<M> for Arc<Noop> {
fn send(&self, _message: M) {}
}
impl<M, R: Send + 'static> CanSendAsync<M, R> for Noop {
fn send_async(&self, _message: M) -> BoxFuture<'static, Result<R, AsyncSendError>> {
async { Err(AsyncSendError::Dropped) }.boxed()
}
}
impl<M, R: Send + 'static> CanSendAsync<M, R> for Arc<Noop> {
fn send_async(&self, _message: M) -> BoxFuture<'static, Result<R, AsyncSendError>> {
async { Err(AsyncSendError::Dropped) }.boxed()
}
}
pub fn noop() -> Arc<Noop> {
Arc::new(Noop)
}
pub trait IntoMultiSender<A> {
fn as_multi_sender(self: &Arc<Self>) -> A;
fn into_multi_sender(self) -> A;
}
pub trait MultiSenderFrom<A> {
fn multi_sender_from(a: Arc<A>) -> Self;
}
impl<A, B: MultiSenderFrom<A>> IntoMultiSender<B> for A {
fn as_multi_sender(self: &Arc<Self>) -> B {
B::multi_sender_from(self.clone())
}
fn into_multi_sender(self) -> B {
B::multi_sender_from(Arc::new(self))
}
}
#[cfg(test)]
mod tests {
use crate::messaging::{AsyncSendError, AsyncSender, CanSendAsync, IntoAsyncSender};
use futures::FutureExt;
use futures::future::BoxFuture;
struct DroppingSender;
impl CanSendAsync<u32, u32> for DroppingSender {
fn send_async(&self, _message: u32) -> BoxFuture<'static, Result<u32, AsyncSendError>> {
async { Err(AsyncSendError::Dropped) }.boxed()
}
}
#[tokio::test]
async fn test_async_sender_from_async_fn_success() {
let sender = AsyncSender::from_fn(|x: u32| x + 1);
let result = sender.send_async(4).await;
assert_eq!(result, Ok(5));
}
#[tokio::test]
async fn test_async_sender_dropped_result() {
let sender = DroppingSender.into_async_sender();
let result = sender.send_async(7).await;
assert_eq!(result, Err(AsyncSendError::Dropped));
}
}