diff --git a/README.md b/README.md index 0e5a150..bc88d5a 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,2 @@ -# about +# swactor Small wasm-compatible actor library diff --git a/examples/hello.rs b/examples/hello.rs index 57f7611..40440e1 100644 --- a/examples/hello.rs +++ b/examples/hello.rs @@ -30,7 +30,7 @@ fn main() { let rt = tokio::runtime::Builder::new_current_thread() .build() .expect("failed to build runtime"); - let greeter = Greeter::spawn(Greeter, &rt); + let greeter = Greeter.spawn(&rt); let response = rt .block_on(async move { diff --git a/src/error.rs b/src/error.rs index 750effe..e00c61c 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1,5 +1,18 @@ -pub type Error = Box; +#[derive(Debug)] +pub struct Error(Box); pub type Result = std::result::Result; -pub fn convert_err(e: E) -> Error { - format!("{e:?}").into() +pub(crate) fn convert_err(e: E) -> Error { + Error(format!("{e:?}").into()) +} + +impl> From for Error { + fn from(value: T) -> Self { + convert_err(value.as_ref()) + } +} + +impl ToString for Error { + fn to_string(&self) -> String { + format!("{:?}", self.0) + } } diff --git a/src/lib.rs b/src/lib.rs index 8bbc7ec..1641435 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,37 +3,64 @@ pub mod error; /// Public export as the oneshot channel is in the `Actor` trait signature pub use tokio::sync::oneshot; -use tokio::{ - sync::{mpsc}, - task::JoinHandle, -}; +use tokio::{sync::mpsc, task::JoinHandle}; use tokio_util::sync::CancellationToken; use crate::error::{Result, convert_err}; - const DEFAULT_CHANNEL_SIZE: usize = 100; +type ActorRequest = ( + ::Message, + oneshot::Sender<::Response>, +); + +/// Wrapper defining the transmission end of a Request/Response channel with an `Actor` +pub struct ActorRequestSender(mpsc::Sender>); + +impl ActorRequestSender { + pub async fn send(&self, request: A::Message) -> Result { + let (tx, rx) = oneshot::channel::(); + self.0.send((request, tx)).await.map_err(convert_err)?; + + rx.await.map_err(convert_err) + } +} + +impl From>> for ActorRequestSender { + fn from(value: mpsc::Sender>) -> Self { + Self(value) + } +} + +impl Clone for ActorRequestSender { + fn clone(&self) -> Self { + Self(self.0.clone()) + } +} + /// Combined 'JoinHandle' to await the actor process and 'Sender' for communication pub struct Handle where A: Actor, { cancel_token: CancellationToken, - tx: mpsc::Sender<(A::Message, oneshot::Sender)>, + tx: ActorRequestSender, + /// the task drops when the `JoinHandle` does, so be careful with the `Handle` _handle: JoinHandle>, // to prevent accidental swaps, strongly type the handle _type: std::marker::PhantomData, } impl Handle { - /// Send a message to the spawned `Actor` task and get a response corresponding to the `Actor::Response` type pub async fn send(&self, msg: A::Message) -> Result { - let (tx, rx) = oneshot::channel::(); - self.tx.send((msg, tx)).await.map_err(convert_err)?; + self.tx.send(msg).await + } - rx.await.map_err(convert_err) + /// Get a cloned sender for messaging the `Actor` this handle is for + pub fn get_connection(&self) -> ActorRequestSender { + self.tx.clone() } } @@ -44,7 +71,7 @@ impl Drop for Handle { } /// Primary trait defining an `Actor` capable of receiving, processing, and transmitting messages -pub trait Actor: Send + Sized + Unpin + 'static { +pub trait Actor: Send + Sized + 'static { /// The type for messages received by this `Actor` type Message: Send; /// The type for responses given by this actor when called from `Handle::send(..)` @@ -84,7 +111,7 @@ pub trait Actor: Send + Sized + Unpin + 'static { Handle { cancel_token, _handle: handle, - tx, + tx: tx.into(), _type: std::marker::PhantomData::, } }