feat: types for request/response handling

This commit is contained in:
Zachery Aaron Shores-Chmielewski 2025-11-25 20:51:13 -05:00
parent 120ba0ec71
commit b8b5b839df
4 changed files with 57 additions and 17 deletions

View file

@ -1,2 +1,2 @@
# about
# swactor
Small wasm-compatible actor library

View file

@ -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 {

View file

@ -1,5 +1,18 @@
pub type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
#[derive(Debug)]
pub struct Error(Box<dyn std::error::Error + Send + Sync + 'static>);
pub type Result<T> = std::result::Result<T, Error>;
pub fn convert_err<E: std::fmt::Debug>(e: E) -> Error {
format!("{e:?}").into()
pub(crate) fn convert_err<E: std::fmt::Debug>(e: E) -> Error {
Error(format!("{e:?}").into())
}
impl<T: AsRef<str>> From<T> for Error {
fn from(value: T) -> Self {
convert_err(value.as_ref())
}
}
impl ToString for Error {
fn to_string(&self) -> String {
format!("{:?}", self.0)
}
}

View file

@ -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<A> = (
<A as Actor>::Message,
oneshot::Sender<<A as Actor>::Response>,
);
/// Wrapper defining the transmission end of a Request/Response channel with an `Actor`
pub struct ActorRequestSender<A: Actor>(mpsc::Sender<ActorRequest<A>>);
impl<A: Actor> ActorRequestSender<A> {
pub async fn send(&self, request: A::Message) -> Result<A::Response> {
let (tx, rx) = oneshot::channel::<A::Response>();
self.0.send((request, tx)).await.map_err(convert_err)?;
rx.await.map_err(convert_err)
}
}
impl<A: Actor> From<mpsc::Sender<ActorRequest<A>>> for ActorRequestSender<A> {
fn from(value: mpsc::Sender<ActorRequest<A>>) -> Self {
Self(value)
}
}
impl<A: Actor> Clone for ActorRequestSender<A> {
fn clone(&self) -> Self {
Self(self.0.clone())
}
}
/// Combined 'JoinHandle' to await the actor process and 'Sender' for communication
pub struct Handle<A>
where
A: Actor,
{
cancel_token: CancellationToken,
tx: mpsc::Sender<(A::Message, oneshot::Sender<A::Response>)>,
tx: ActorRequestSender<A>,
/// the task drops when the `JoinHandle` does, so be careful with the `Handle`
_handle: JoinHandle<Result<()>>,
// to prevent accidental swaps, strongly type the handle
_type: std::marker::PhantomData<A>,
}
impl<A: Actor> Handle<A> {
/// 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<A::Response> {
let (tx, rx) = oneshot::channel::<A::Response>();
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<A> {
self.tx.clone()
}
}
@ -44,7 +71,7 @@ impl<A: Actor> Drop for Handle<A> {
}
/// 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::<Self>,
}
}