diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e9baa92..f1d85d5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,8 +19,8 @@ jobs: with: rust-version: nightly components: rustfmt - - uses: actions/checkout@v4 - - run: cargo fmt --check + - uses: actions/checkout@v7 + - run: cargo +nightly fmt --check wasm_build: needs: code_cleaniness_checks name: Wasm Tests @@ -33,7 +33,9 @@ jobs: with: rust-version: ${{ matrix.rust }} targets: wasm32-unknown-unknown - - uses: actions/checkout@v4 + - name: Cache Rust dependencies + uses: Swatinem/rust-cache@v2 + - uses: actions/checkout@v7 - run: cargo build --target=wasm32-unknown-unknown --verbose - run: curl https://rustwasm.github.io/wasm-pack/installer/init.sh -sSf | sh - run: wasm-pack test --headless --chrome --firefox --verbose @@ -51,7 +53,9 @@ jobs: with: rust-version: ${{ matrix.rust }} components: rustfmt - - uses: actions/checkout@v4 + - name: Cache Rust dependencies + uses: Swatinem/rust-cache@v2 + - uses: actions/checkout@v7 - run: cargo fmt --check - run: cargo build --verbose - run: cargo test --verbose @@ -62,8 +66,10 @@ jobs: name: Coverage runs-on: ubuntu-latest steps: + - name: Cache Rust dependencies + uses: Swatinem/rust-cache@v2 - name: Checkout repository - uses: actions/checkout@v4 + uses: actions/checkout@v7 - name: Setup Rust uses: hecrj/setup-rust-action@v2 with: diff --git a/Cargo.toml b/Cargo.toml index e356aa9..d478387 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,6 @@ tokio-stream = { version = "0.1", features = ["sync"] } futures = "0.3.19" instant = { version = "0.1" } pin-project = { version = "1.1" } -anymap2 = "0.13" [target.'cfg(target_arch = "wasm32")'.dependencies] wasm-bindgen-futures = { version = "0.4.37" } diff --git a/README.md b/README.md index da62f60..7eeb11a 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,9 @@ Do note that actor communication uses `tokio`'s channels regardless of the async The heart of a `troupe` actor is a state type and a message type. Let's take a simple cache as an example: ```rust -pub struct Cache(HashMap); +pub struct Cache { + inner: HashMap, +} pub enum CacheCommand { Insert(Uuid, YourData), @@ -40,11 +42,8 @@ pub enum CacheCommand { The message type encapsulates how the state can change and what data the actor has to process. These messages are sent from various source to be processed by the state. To finish the actor, `Cache` just has to implement the `ActorState` trait. ```rust impl ActorState for Cache { - type Permanence = Permanent; - type ActorType = SinkActor; - + type ActorKind = SinkActor; type Message = CacheCommand; - type Output = (); async fn process(&mut self, _: &mut Scheduler, msg: CacheCommand) { match msg { @@ -53,7 +52,7 @@ impl ActorState for Cache { } CacheCommand::Get(key, send) => { let _ = send.send(self.inner.get(&key).cloned()); - } + } CacheCommand::Delete(key) => { self.inner.remove(&key); } @@ -62,18 +61,28 @@ impl ActorState for Cache { } ``` -`ActorState` has several other associated types beyond the message type. -These types are mostly marker types and inform how other parts of your program should interact with the actor. -This communication is done through a client, which is created when the actor is launched. -The associated `ActorType` type tells the client if the actor expects messages to be sent to it or if messages will be broadcast from the actor (or both). -The associated `Permanence` type informs the client if it should expect the actor to close at any point. -Lastly, the associated `Output` type is only used for actor that broadcast messages, in which case the actor will broadcast messages of the `Output` type. +Beyond the message type, `ActorState` has one other associated type: its `ActorKind`. +Where the state describes how the actor reacts to messages, the kind describes how messages get into and out of the actor. +`troupe` ships three kinds: `SinkActor` for actors that only receive messages, `StreamActor` for actors that only broadcast them, and `JointActor` for actors that do both. + +The kind is not just a label. +When the actor is launched, its kind is constructed, and that construction does three things. +It builds the client that is handed back to the caller, it attaches whatever inbound streams that client needs (such as the MPSC channel behind a `SinkClient`), and it holds whatever state the actor needs in order to send messages back out (such as the broadcast channel behind a `StreamClient`). +That last piece is stored in the scheduler, which derefs to the kind, so a stream or joint actor broadcasts by simply calling `scheduler.broadcast(msg)`. + +A kind can also require configuration via its associated `Config` type, which is handed to the builder with `ActorBuilder::config` before launching. +All three of the kinds above use `()` here, so they can be launched directly. + +Because `ActorKind` is a public trait, other message-passing patterns can be modelled without changing the scheduler or the builder. +Clients with built-in back pressure or an SPSC-style client, for example, are each just a new kind and its client type. Once running, `troupe` pairs every actor state with a scheduler. The scheduler is responsible for managing futures that the state queues and attached streams of message. The queued futures will be polled at the same time that the scheduler waits for inbound messages. -Most actors have one attached stream by default, the channel used to communicate between the client and the actor. -Client message streams use tokio MPSC-style channels, but actors can add any stream that yield messages that can be converted into the actor's message type (such as socket connections). +Sink and joint actors start out with one attached stream, the channel used to communicate between the client and the actor, which their `ActorKind` attaches at launch. +Client message streams use tokio MPSC-style channels, but actors can add any stream that yield messages that can be converted into the actor's message type (such as socket connections), either up front with `ActorBuilder::attach_stream` or at any point afterwards with `Scheduler::attach_stream`. +A stream actor has no client-backed channel, so all of its messages come from streams and futures added this way. +The scheduler also holds the actor's kind and derefs to it, which is how a state reaches methods like `broadcast`. Conceptually, the scheduler-actor relationship can be model as: ``` _____________________________________________ @@ -99,18 +108,13 @@ Unlike a `SinkClient`, a `StreamClient` does not directly support any type of me It does, however, implement the [`Stream`](https://docs.rs/futures/latest/futures/stream/trait.Stream.html) trait. Lastly, a `JointClient` is both a `SinkClient` and a `StreamClient` put together. A `JointClient` can be decomposed into one or both of the other clients. -Note, the actor type defines what kind of clients can be constructed, so you can not, for example, construct a `StreamClient` for an actor that will never broadcast a message. - -The last major component of the `troupe` actor model is permanence. -Some actors are designed to run forever. -These are called `Permanent` actors. -Other actors are designed to run for some time (perhaps a long time) and then close. -These are called `Transient` actors. -This distinction largely serves to help with the ergonomics of interacting with actors. -If an actor is designed to never shutdown, then the oneshot channels used for request-response style messages can be safely unwrapped. -The same is not true for `Transient` actors. - -Regardless of the permanence of the actor, all actors might exhaust their source of messages. +Note, the actor's kind defines what kind of client is constructed, so you can not, for example, get a `StreamClient` for an actor that will never broadcast a message. + +Actors do not run forever. +An actor can retire itself at any point by calling `Scheduler::shutdown` (or `Scheduler::shutdown_and_finish`, which lets the futures the scheduler is managing for it run to completion first). +Because of this, the response to a request-response style message is not guaranteed to arrive, so awaiting a `Tracker` yields an `Option`. + +Actors might also exhaust their source of messages. This happens when all streams have ran dry and no message-yeilding futures are queued. When this happens, there is built-in "garbage collection" for troupe actors. The scheduler will mark an actor as "dead" and then shutdown the actor process if it ever reaches this state. @@ -119,14 +123,13 @@ For `SinkActors`, this can only occur if all message-sending clients have been d # Backwards Compatibility Troupe is currently experimental and subject to potential breaking changes (with due consideration). Breaking changes might occur to improve API ergonomics, to tweak the actor model, or to use new Rust language features. -In particular, there are several language features that will be used improve this crate upon stabilization: - - [`async fn` in traits](https://rust-lang.github.io/async-book/07_workarounds/05_async_in_traits.html) - - [specialization](https://rust-lang.github.io/rfcs/1210-impl-specialization.html) - - [associate-type defaults](https://rust-lang.github.io/rfcs/2532-associated-type-defaults.html) - -The stabilization of the first two present garunteed breaking changes for this crate but will drastically improve usability and ergonomics. -Specialization will enable the `ActorBuilder` to present identically-named methods for launching the actor while returning the appropiate client type. -The stabilization of `async fn` in traits will allow for the loosening of constraints on `ActorState`s in WASM contexts, allowing them to just be `'static` instead of `'static + Send`. +In particular, there are still language features that would improve this crate upon stabilization: + - [async fn in traits with `Send` bounds](https://rust-lang.github.io/rfcs/3654-return-type-notation.html) + - [associated-type defaults](https://rust-lang.github.io/rfcs/2532-associated-type-defaults.html) + +`ActorState`'s methods are written as `-> impl MaybeSendFuture` rather than as plain `async fn` so that the `Send` bound can be dropped on WASM targets. +A stable way to write a maybe-`Send` `async fn` in a trait would let that workaround, and the `Sendable` bound it carries, be removed. +Associated-type defaults would let `ActorKind::Config` default to `()`, so kinds that need no configuration would not have to spell it out. # Future Work Currently, `troupe` only provides a framework for building actors. diff --git a/examples/cache/Cargo.lock b/examples/cache/Cargo.lock index 660f274..dde6592 100644 --- a/examples/cache/Cargo.lock +++ b/examples/cache/Cargo.lock @@ -1,6 +1,6 @@ # This file is automatically @generated by Cargo. # It is not intended for manual editing. -version = 3 +version = 4 [[package]] name = "addr2line" @@ -17,17 +17,6 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe" -[[package]] -name = "async-trait" -version = "0.1.74" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a66537f1bb974b254c98ed142ff995236e81b9d0fe4db0575f46612cb15eb0f9" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.39", -] - [[package]] name = "autocfg" version = "1.1.0" @@ -67,6 +56,15 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2bd12c1caf447e69cd4528f47f94d203fd2582878ecb9e9465484c4148a8223" +[[package]] +name = "cache" +version = "0.1.0" +dependencies = [ + "derive_more", + "tokio", + "troupe", +] + [[package]] name = "cc" version = "1.0.83" @@ -132,15 +130,6 @@ version = "0.3.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eb1d22c66e66d9d72e1758f0bd7d4fd0bee04cad842ee34587d68c07e45d088c" -[[package]] -name = "futures-example-imperative" -version = "0.1.0" -dependencies = [ - "derive_more", - "tokio", - "troupe", -] - [[package]] name = "futures-executor" version = "0.3.29" @@ -548,9 +537,8 @@ dependencies = [ [[package]] name = "troupe" -version = "0.1.0" +version = "0.2.0" dependencies = [ - "async-trait", "futures", "gloo-timers", "instant", diff --git a/examples/cache/Cargo.toml b/examples/cache/Cargo.toml index ac992eb..e187524 100644 --- a/examples/cache/Cargo.toml +++ b/examples/cache/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "futures-example-imperative" +name = "cache" edition = "2021" version = "0.1.0" publish = false diff --git a/examples/cache/src/cache.rs b/examples/cache/src/cache.rs index 508ba73..36fb022 100644 --- a/examples/cache/src/cache.rs +++ b/examples/cache/src/cache.rs @@ -21,20 +21,16 @@ pub enum CacheCommand { Query(K, Box)>), } -#[async_trait] impl ActorState for Cache where K: 'static + Send + Hash + Eq, T: 'static + Send + Clone, { - type Message = CacheCommand; - - /// This actor is a [`SinkActor`] as it does not broadcast anything - type ActorType = SinkActor; + /// This actor is a [`SinkActor`] as it does not broadcast anything. That kind gives us a + /// [`SinkClient`] when the actor is launched. + type ActorKind = SinkActor; - /// Sink actors don't output anything, so we can use a unit here. Ideally, [`ActorState`] would - /// have a type default for this. - type Output = (); + type Message = CacheCommand; async fn process(&mut self, _: &mut Scheduler, msg: CacheCommand) { println!("Message received!!"); diff --git a/examples/cache/src/main.rs b/examples/cache/src/main.rs index 6aa9564..5a9572c 100644 --- a/examples/cache/src/main.rs +++ b/examples/cache/src/main.rs @@ -6,15 +6,35 @@ mod cache; #[tokio::main] async fn main() { let state: Cache = Cache::default(); - let client = ActorBuilder::new(state).launch_sink(); + // `SinkActor`'s config is `()`, so the builder can be launched directly. Launching hands back + // the kind's client, a `SinkClient`. + let client = ActorBuilder::new(state).launch(); client.send((1, "one")); - let val = client.track(1).await.unwrap(); + // Awaiting a `Tracker` yields an `Option` because the actor may have shut down before it could + // respond. Here, the actor's response is itself an `Option`, hence the `flatten`. + let val = client.track(1).await.flatten().unwrap(); + println!("Fetched: {val:?}"); assert_eq!(val, "one"); - let answer = client.track((1, |s: &&str| s.to_string())).await.unwrap_or_default(); + let answer = client + .track((1, |s: &&str| s.to_string())) + .await + .flatten() + .unwrap_or_default(); + println!("Fetched: {answer:?}"); assert_eq!(answer, "one"); - let answer = client.track((2, |s: &&str| s.to_string())).await.unwrap_or_default(); + let answer = client + .track((2, |s: &&str| s.to_string())) + .await + .flatten() + .unwrap_or_default(); + println!("Fetched: {answer:?}"); assert_eq!(answer, ""); client.send(1); - let answer = client.track((1, |s: &&str| s.to_string())).await.unwrap_or_default(); + let answer = client + .track((1, |s: &&str| s.to_string())) + .await + .flatten() + .unwrap_or_default(); + println!("Fetched: {answer:?}"); assert_eq!(answer, ""); } diff --git a/examples/ping/Cargo.lock b/examples/ping/Cargo.lock index 660f274..099d235 100644 --- a/examples/ping/Cargo.lock +++ b/examples/ping/Cargo.lock @@ -1,6 +1,6 @@ # This file is automatically @generated by Cargo. # It is not intended for manual editing. -version = 3 +version = 4 [[package]] name = "addr2line" @@ -17,17 +17,6 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe" -[[package]] -name = "async-trait" -version = "0.1.74" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a66537f1bb974b254c98ed142ff995236e81b9d0fe4db0575f46612cb15eb0f9" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.39", -] - [[package]] name = "autocfg" version = "1.1.0" @@ -132,15 +121,6 @@ version = "0.3.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eb1d22c66e66d9d72e1758f0bd7d4fd0bee04cad842ee34587d68c07e45d088c" -[[package]] -name = "futures-example-imperative" -version = "0.1.0" -dependencies = [ - "derive_more", - "tokio", - "troupe", -] - [[package]] name = "futures-executor" version = "0.3.29" @@ -372,6 +352,15 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +[[package]] +name = "ping" +version = "0.1.0" +dependencies = [ + "derive_more", + "tokio", + "troupe", +] + [[package]] name = "proc-macro2" version = "1.0.69" @@ -548,9 +537,8 @@ dependencies = [ [[package]] name = "troupe" -version = "0.1.0" +version = "0.2.0" dependencies = [ - "async-trait", "futures", "gloo-timers", "instant", diff --git a/examples/ping/Cargo.toml b/examples/ping/Cargo.toml index ac992eb..b66cc87 100644 --- a/examples/ping/Cargo.toml +++ b/examples/ping/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "futures-example-imperative" +name = "ping" edition = "2021" version = "0.1.0" publish = false diff --git a/examples/ping/src/main.rs b/examples/ping/src/main.rs index ffa07ae..59bc72b 100644 --- a/examples/ping/src/main.rs +++ b/examples/ping/src/main.rs @@ -7,28 +7,32 @@ mod ping; #[tokio::main] async fn main() { - // Construct the builder for the ping "server" - let builder = ActorBuilder::new(Ping::default()); - // We can get a client before launching the actor process. - let client = builder.sink_client(); - // We can send messages to the client but they will not be processed until we launch the actor. + // Spawn the actor process. `SinkActor`'s config is `()`, so the builder needs nothing else + // before launching. Launching hands back the kind's client, a `SinkClient`. + let client = ActorBuilder::new(Ping::default()).launch(); + // The actor processes messages in the order that they are sent, so we can queue up a series of + // messages and know what each response will be. client.send(()); - let tracker = client.track(()); + let first = client.track(()); + let second = client.track(()); + // Clients are cheap to clone, so each part of your application can hold its own. + let other_client = client.clone(); tokio::spawn(async move { - let resp = tracker.await; + // Awaiting a `Tracker` yields an `Option` because the actor may have shut down before it + // could respond. + let resp = first.await.unwrap(); println!("Got response from ping server: {resp}"); - // Messages are processed in order, so we know that the second message will yield 1 + // The first `send` bumped the counter, so this message sees a count of 1. assert_eq!(resp, 1); }); tokio::spawn(async move { - let resp = client.track(()).await; + let resp = second.await.unwrap(); println!("Got response from ping server: {resp}"); assert_eq!(resp, 2); + drop(other_client); }); - // Spawn the actor process, which returns a client - let client = builder.launch_sink(); tokio::time::sleep(Duration::from_millis(10)).await; - let resp = client.track(()).await; + let resp = client.track(()).await.unwrap(); println!("Got response from ping server: {resp}"); assert_eq!(resp, 3); } diff --git a/examples/ping/src/ping.rs b/examples/ping/src/ping.rs index dde3357..c1cc274 100644 --- a/examples/ping/src/ping.rs +++ b/examples/ping/src/ping.rs @@ -18,19 +18,15 @@ pub enum PingCommand { RequestAndRespond(OneshotSender), } -#[async_trait] impl ActorState for Ping { + /// This actor is a [`SinkActor`] as it does not broadcast anything. That kind gives us a + /// [`SinkClient`] when the actor is launched. + type ActorKind = SinkActor; + /// A message of type `T` must be `T: Into` or `(T, OneshotSender): /// Into, msg: PingCommand) { println!("Ping message receive: {msg}"); match msg { diff --git a/rustfmt.toml b/rustfmt.toml new file mode 100644 index 0000000..63df91e --- /dev/null +++ b/rustfmt.toml @@ -0,0 +1 @@ +imports_granularity = "Item" diff --git a/src/compat/mod.rs b/src/compat/mod.rs index 39dcdc2..a6482eb 100644 --- a/src/compat/mod.rs +++ b/src/compat/mod.rs @@ -2,7 +2,9 @@ use std::fmt::Debug; -use futures::{stream::FusedStream, Future, Stream}; +use futures::stream::FusedStream; +use futures::Future; +use futures::Stream; #[cfg(not(target_family = "wasm"))] mod native; @@ -55,7 +57,13 @@ impl Debug for Sleep { mod test { // Import tests #[allow(unused_imports)] - use super::{sleep_for, sleep_until, spawn_task, Sendable}; + use super::sleep_for; + #[allow(unused_imports)] + use super::sleep_until; + #[allow(unused_imports)] + use super::spawn_task; + #[allow(unused_imports)] + use super::Sendable; // Impl trait tests } diff --git a/src/compat/native.rs b/src/compat/native.rs index a62ee92..b92318e 100644 --- a/src/compat/native.rs +++ b/src/compat/native.rs @@ -1,5 +1,3 @@ -use anymap2::any::Any; - /* ------ Send workarounds ------ */ /// This trait abstracts over put of the requirements for spawning an async task. In native async runtimes, a @@ -11,8 +9,6 @@ pub trait MaybeSend: Send {} impl MaybeSend for T where T: Send {} -pub(crate) type SendableAnyMap = anymap2::Map; - /* ------ General Utils ------ */ #[cfg(all(feature = "tokio", feature = "async-std"))] @@ -27,13 +23,13 @@ pub use async_std::*; #[cfg(feature = "tokio")] mod tokio { use crate::compat::SendableFuture; - use instant::{Duration, Instant}; + use instant::Duration; + use instant::Instant; use pin_project::pin_project; - use std::{ - future::Future, - pin::Pin, - task::{Context, Poll}, - }; + use std::future::Future; + use std::pin::Pin; + use std::task::Context; + use std::task::Poll; use super::super::Sendable; @@ -75,12 +71,12 @@ mod tokio { mod async_std { use crate::compat::SendableFuture; use futures::FutureExt; - use instant::{Duration, Instant}; - use std::{ - future::Future, - pin::Pin, - task::{Context, Poll}, - }; + use instant::Duration; + use instant::Instant; + use std::future::Future; + use std::pin::Pin; + use std::task::Context; + use std::task::Poll; use super::super::Sendable; diff --git a/src/compat/wasm.rs b/src/compat/wasm.rs index 775c4ea..30bf7ea 100644 --- a/src/compat/wasm.rs +++ b/src/compat/wasm.rs @@ -1,14 +1,15 @@ -use anymap2::any::Any; use futures::FutureExt; -use gloo_timers::future::{sleep, TimeoutFuture}; -use instant::{Duration, Instant}; -use std::{ - future::Future, - pin::Pin, - task::{Context, Poll}, -}; - -use super::{Sendable, SendableFuture}; +use gloo_timers::future::sleep; +use gloo_timers::future::TimeoutFuture; +use instant::Duration; +use instant::Instant; +use std::future::Future; +use std::pin::Pin; +use std::task::Context; +use std::task::Poll; + +use super::Sendable; +use super::SendableFuture; /* ------ Send workarounds ------ */ @@ -21,8 +22,6 @@ pub trait MaybeSend {} impl MaybeSend for T {} -pub(crate) type SendableAnyMap = anymap2::Map; - /* ------ General Utils ------ */ /// A wrapper around `wasm-bindgen-future`'s `spawn_local` function, which spawns a future tha diff --git a/src/joint.rs b/src/joint.rs index 58fc797..79f5da9 100644 --- a/src/joint.rs +++ b/src/joint.rs @@ -1,54 +1,97 @@ //! Actors that both can be sent messages and broadcast messages. -use std::{ - pin::Pin, - task::{Context, Poll}, -}; +use std::pin::Pin; +use std::task::Context; +use std::task::Poll; use futures::Stream; use pin_project::pin_project; +use tokio::sync::broadcast; +use tokio::sync::oneshot; + +use crate::compat::Sendable; +use crate::sink::SinkActor; +use crate::sink::SinkClient; +use crate::sink::Tracker; +use crate::stream::Broadcastee; +use crate::stream::StreamActor; +use crate::stream::StreamClient; +use crate::ActorKind; +use crate::ActorState; +use crate::Scheduler; + +/// The [`ActorKind`] for actors that both receive and broadcast messages. A joint actor is one +/// that acts as both a [`SinkActor`] and a [`StreamActor`]. Its clients, [`JointClient`]s, can +/// both send messages into the actor and recieve messages forwarded by the actor. +/// +/// Note that the actor's inbound and outbound message types are distinct. Inbound messages are the +/// state's [`Message`](ActorState::Message); outbound messages are this kind's `M`. +/// +/// Constructing this kind does the work of both of its halves: it creates the MPSC channel that +/// backs the client's sink side and attaches the receiving end to the [`Scheduler`], and it holds +/// the broadcast sender that backs the client's stream side. Since the kind lives in the +/// scheduler, which derefs to it, an [`ActorState`] broadcasts by calling +/// [`broadcast`](JointActor::broadcast) on the scheduler it is given. +#[derive(Debug)] +pub struct JointActor { + broadcast: broadcast::Sender>, +} -use crate::{ - compat::Sendable, - sink::{self, SinkClient}, - stream::StreamClient, - Permanent, Transient, -}; +impl ActorKind for JointActor { + type Client = JointClient; + type Config = (); -use crate::OneshotSender; + fn construct((): Self::Config) -> (Self, Self::Client, impl FnOnce(&mut Scheduler)) { + let (SinkActor {}, send, init_one) = SinkActor::construct(()); + let (StreamActor { broadcast }, recv, init_two) = StreamActor::construct(()); -/// A marker type used by the [`ActorBuilder`](crate::ActorBuilder) to know what kind of -/// [`ActorState`](crate::ActorState) it is dealing with. A joint actor is one that acts as both a -/// [`SinkActor`](crate::sink::SinkActor) and a [`StreamActor`](crate::stream::StreamActor). Its -/// clients, [`JointClient`]s, can both send messages into the actor and recieve messages forwarded -/// by the actor. -#[derive(Debug)] -pub struct JointActor; + let this = Self { broadcast }; + let client = JointClient { send, recv }; + let init = move |scheduler: &mut Scheduler| { + init_one(scheduler); + init_two(scheduler); + }; + + (this, client, init) + } +} + +impl JointActor { + /// Broadcasts a message to all listening clients. If the message fails to send, the message + /// will be dropped. + /// + /// This is normally reached through the [`Scheduler`], which derefs to this type, rather than + /// on a `JointActor` directly. + pub fn broadcast(&mut self, msg: impl Into) { + #[cfg(not(target_family = "wasm"))] + let _ = self.broadcast.send(msg.into()); + #[cfg(target_family = "wasm")] + let _ = self + .broadcast + .send(send_wrapper::SendWrapper::new(msg.into())); + } +} /// A client to an actor. This client is a combination of the [`SinkClient`] and the -/// [`StreamClient`]. +/// [`StreamClient`]. `I` is the type of message sent *into* the actor and `O` is the type of +/// message broadcast *out* of it. #[pin_project] #[derive(Debug)] -pub struct JointClient { - send: SinkClient, +pub struct JointClient { + send: SinkClient, #[pin] recv: StreamClient, } -impl JointClient { - /// A constuctor for the client. - pub(crate) fn new(send: SinkClient, recv: StreamClient) -> Self { - Self { send, recv } - } - +impl JointClient { /// Consumes the client and return the constituent sink and stream clients. - pub fn split(self) -> (SinkClient, StreamClient) { + pub fn split(self) -> (SinkClient, StreamClient) { let Self { send, recv } = self; (send, recv) } /// Returns a clone of this client's sink client. - pub fn sink(&self) -> SinkClient { + pub fn sink(&self) -> SinkClient { self.send.clone() } @@ -70,40 +113,19 @@ impl JointClient { pub fn send(&self, msg: impl Into) -> bool { self.send.send(msg) } -} - -impl JointClient { - /// Sends a request-response style message to a [`Permanent`] actor. The given data is paired - /// with a one-time use channel and sent to the actor. A - /// [`Tracker`](crate::sink::permanent::Tracker) that will receive a response from the actor is - /// returned. - /// - /// Note: Since this client is one for a permanent actor, there is an implicit unwrap once the - /// tracker receives a message from the actor. If the actor drops the other half of the channel - /// or has died somehow (likely from a panic), the returned tracker will panic too. So, it is - /// important that the actor always sends back a message - pub fn track(&self, msg: M) -> sink::permanent::Tracker - where - I: From<(M, OneshotSender)>, - { - self.send.track(msg) - } -} -impl JointClient { - /// Sends a request-response style message to a [`Transient`] actor. The given data is paired - /// with a one-time use channel and sent to the actor. A - /// [`Tracker`](crate::sink::transient::Tracker) that will receive a response from the actor is - /// returned. - pub fn track(&self, msg: M) -> sink::transient::Tracker + /// Sends a request-response style message to an actor. The given data is paired with a + /// one-time use channel and sent to the actor. A [`Tracker`] that will receive a response from + /// the actor is returned. + pub fn track(&self, msg: M) -> Tracker where - I: From<(M, OneshotSender)>, + I: From<(M, oneshot::Sender)>, { self.send.track(msg) } } -impl Clone for JointClient +impl Clone for JointClient where O: Sendable + Clone, { @@ -115,7 +137,7 @@ where } } -impl Stream for JointClient +impl Stream for JointClient where O: Sendable + Clone, { diff --git a/src/lib.rs b/src/lib.rs index 5b6e526..49ab4f0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,30 +1,57 @@ -//! Troupe provides a high-level toolset for modelling crates with actors. Troupe actors are built -//! on top of async process, like those created from `tokio::spawn`, and help you model and control -//! the flow of information in and out of them. The main goals of `troupe` are to provide: +//! Troupe provides a high-level toolset for modelling states and data flows. Troupe actors are +//! built on top of async tasks (like those created from [`tokio::spawn`](tokio::spawn)) and help +//! you model and control the flow of information into and out of them. The main goals of `troupe` +//! are to provide: +//! - Consise ways of modelling state and state changes +//! - APIs that remove the boilerplate of creating and running async actors //! - An easy to conceptualize data flow -//! - Simple access to concurrently processing of futures within actors +//! - Ergonomic APIs devoid of magic //! - A model that can be adopted into an existing project all at once or over time -//! - An ergonomic API devoid of magic +//! - Support for both native and WASM targets //! -//! At the core of every actor is an [`ActorState`]. These are the building blocks used to model -//! your program with `troupe`. This state is fully isolated from the rest of your application and -//! can only be reached by attaching a stream of messages. For many actors, a stream is provided by -//! `troupe` in the form of an [`mpsc-style`](tokio::sync::mpsc) tokio channel. All attached -//! streams are managed by a [`Scheduler`]. The state can attach new streams, queue futures that -//! yield message, or hand off futures that yield nothing to the scheduler. +//! Troupe currently supports three async runtimes: `tokio`, `async-std`, and the runtime provided +//! by the browser (via wasm-bindgen-futures). Do note that even if you are using the `async-std` +//! runtime, client-actor communication is still done via tokio channels. //! -//! Communication to and from an actor is managed by a client. Each actor state defines how its -//! clients should function via the [`ActorState`]'s `ActorType`. Conceptually, every actor is -//! either something that consumes messages, i.e. a `Sink`, or something to broadcasts messages, -//! i.e. a `Stream`, or both. An actor that largely receive messages from other parts of our -//! program is a [`SinkActor`], which use [`SinkClient`]s. An actor that processes messages from a -//! source (for example a Websocket) and then broadcast these messages is a [`StreamActor`], which -//! use [`StreamClient`]s. If an actor does both of these, it is a [`JointActor`] and uses -//! [`JointClient`]s. +//! The general model of an actor is a state machine that receieves messages from surrounding +//! state, updates its state, and potentially sends messages to other actors and/or spawns more +//! actors (see the [wiki](https://en.wikipedia.org/wiki/Actor_model)). Troupe handles all of the +//! plumbing so you can focus on modelling your state. The state of an actor is modelled with the +//! [`ActorState`] trait. //! -//! Troupe currently supports three async runtimes: `tokio`, `async-std`, and the runtime provided by -//! the browser (via wasm-bindgen-futures). Do note that even if you are using the `async-std` -//! runtime, client-actor communication is still done via tokio channels. +//! While `troupe` handles all of the plumbing for you, knowing how messages are sent and receieved +//! is key to leveraging `troupe` to its full potential. Messages can have +//! +//! Messages arriving at the actors, messages the actor sends itself, messages leaving the actors, +//! +//! Imagine a simple case where, in a web server, an actor receives messages that need to be logged +//! for telemetry. Troupe provides mechanisms for attaching [`Stream`]s to an actor's task. An +//! actor can have any number of [`Stream`]s attached, and all of them are selected on +//! indescrimiately. Meaning if two streams should yeild items one after another, that invariant is +//! not enforced by `troupe`. +//! +//! Most actors have one stream by default, their client. When spawned, a client to the actor is +//! returned. Clients can be modelled in many ways, but many provide a way to send messages into +//! the actor. In the logging actor example, the state for the rest of the server would have +//! clients to the logging actor. Those clients would then send data that could then be logged. +//! +//! Actors can receive messages from places other than their client(s). Consider that same web +//! server. Perhaps it is using [`websockets`]() somewhere in its stack. A websocket is, +//! effectively, just a [`Stream`] of messages from one of the machine's socket. Troupe provides an +//! [`ActorState`] with the ability to attach a websocket, as a [`Stream`], so the actor respond to +//! messages from the websocket just like it would respond to client messages. Similarly, perhaps +//! that actor also needs to send an HTTP request. That request can be awaited and the resulting +//! response turned into a message that the actor can process. All of this is handled by the +//! [`Scheduler`]. +//! +//! NOTE: Since all [`Stream`]s *and* [`Future`]s handled by the [`Scheduler`] are selected on, +//! they need to be cancel safe. + +// TODO: +// - Make ActorState::start_up return V +// - Normalize the language between docs and across methods, e.g. "ActorBuilder::launch" conflicts +// with "spawning an actor" +// - Make Unbound/backpressure clients #![warn(rust_2018_idioms)] #![deny( @@ -53,33 +80,33 @@ pub(crate) mod scheduler; pub mod sink; pub mod stream; -use compat::{MaybeSendFuture, Sendable, SendableAnyMap, SendableFusedStream}; -use joint::{JointActor, JointClient}; -pub use scheduler::Scheduler; -use scheduler::{ActorRunner, ActorStream}; -#[cfg(target_family = "wasm")] -use send_wrapper::SendWrapper; -use sink::{SinkActor, SinkClient}; -use stream::{StreamActor, StreamClient}; -pub use tokio::sync::oneshot::{ - channel as oneshot_channel, Receiver as OneshotReceiver, Sender as OneshotSender, -}; +use futures::StreamExt; -use std::marker::PhantomData; +use compat::MaybeSendFuture; +use compat::Sendable; +use compat::SendableFusedStream; +use scheduler::ActorRunner; +use scheduler::ActorStream; -use futures::StreamExt; -use tokio::sync::{ - broadcast, - mpsc::{unbounded_channel, UnboundedSender}, -}; +pub use scheduler::Scheduler; + +#[cfg(doc)] +use prelude::*; +#[cfg(doc)] +use sink::*; +#[cfg(doc)] +use stream::*; /// The core abstraction of the actor model. An [`ActorState`] sits at the heart of every actor. It /// processes messages, queues futures, and attaches streams in the [`Scheduler`], and it can -/// forward messages. Actors serves two roles. They can act similarly to a +/// forward messages. +/// +/// +/// Actors serves two roles. They can act similarly to a /// [`Sink`](futures::Sink) where other parts of your application (including other actors) since /// messages into the actor. They can also act as a [`Stream`](futures::Stream) that generate /// messages to be sent throughout your application. This role is denoted by the actor's -/// `ActorType`, which informs the [`ActorBuilder`] what kind of actor it is working with. For +/// [`ActorKind`], which informs the [`ActorBuilder`] what kind of actor it is working with. For /// sink-like actors, use the [`SinkActor`] type. For stream-like actors, use the [`StreamActor`] /// type. For actors that function as both, use the [`JointActor`] type. /// @@ -87,27 +114,17 @@ use tokio::sync::{ /// fn` to implement them. Their current bounds of `MaybeSendFuture` are there to abstract over the /// different requirements for native and WASM targets. pub trait ActorState: Sendable + Sized { - /// This type should either be [`SinkActor`], [`StreamActor`], or [`JointActor`]. This type is - /// mostly a marker to inform the [`ActorBuilder`]. - type ActorType; - - /// This type should either be [`Permanent`] or [`Transient`]. This type is mostly a marker - /// type to inform the actor's client(s) if it should expect the actor to shutdown at any - /// point. - type Permanence; - - /// Inbound messages to the actor must be this type. Clients will send the actor messages of - /// this type and any queued futures or streams must yield this type. + /// The type that inbound messages to the actor will be. type Message: Sendable; - /// For [`SinkActor`]s and [`JointActor`]s, this is the message type which is broadcasts. - /// For [`StreamActor`]s, this can be `()` (unfortunately, default associated types are - /// unstable). - type Output: Sendable + Clone; + /// Actors can serve different roles in the surrounding state, defined by the type of client + /// the actor has. The [`ActorKind`] defines those clients as well as additional state the + /// [`Scheduler`] has to interact with those clients. + type ActorKind: ActorKind; /// Before starting the main loop of running the actor, this method is called to finalize any - /// setup of the actor state, such as pulling data from a database or from over the network. No - /// inbound messages will be processed until this method is completed. + /// setup of the actor state, such as pulling data from a database or from over the network or + /// simply waiting for the first few messages from the [`Scheduler`] /// /// Note: When implementing this method, you can use `async fn` instead of `impl /// MaybeSendFuture`. @@ -116,11 +133,11 @@ pub trait ActorState: Sendable + Sized { std::future::ready(()) } - /// The heart of the actor. This method consumes messages attached streams and queued futures - /// and streams. For [`SinkActor`]s and [`JointActor`]s, the state can "respond" to messages - /// containing a [`OneshotChannel`](tokio::sync::oneshot::channel) sender. The state can also - /// queue futures and attach streams in the [`Scheduler`]. Finally, for [`StreamActor`]s and - /// [`JointActor`]s, the state can broadcast messages via [`Scheduler`]. + /// The heart of the actor. This method consumes messages yielded by attached streams and + /// queued futures. As part of processing a message, an actor can spawn other actors, provide + /// streams and futures to the [`Scheduler`] to be managed, send messages other actors via + /// clients the state has, and, if applicable, interact with outbound streams provided by the + /// actor's [`ActorKind`]. /// /// Note: When implementing this method, you can use `async fn` instead of `impl /// MaybeSendFuture`. @@ -130,9 +147,8 @@ pub trait ActorState: Sendable + Sized { msg: Self::Message, ) -> impl MaybeSendFuture; - /// Once the actor has died, this method is called to allow the actor to clean up anything that - /// remains. Note that this method is also called even for [`Permanent`] actors that have - /// expired. + /// Once the actor has completed, this method is called to allow the actor to clean up anything + /// that remains. /// /// Note: When implementing this method, you can use `async fn` instead of `impl /// MaybeSendFuture`. @@ -142,206 +158,117 @@ pub trait ActorState: Sendable + Sized { } } -/// A marker type used in the [`ActorState`]. It communicates that the actor should never die. As -/// such, the [`Scheduler`] will not provide the actor state a method to shutdown. Also, the -/// [`Tracker`](crate::sink::permanent::Tracker)s for request-response style messages will implictly unwrap responses from their -/// oneshot channels. -#[derive(Debug)] -pub struct Permanent; +/// Where [`ActorState`] describes how an actor reacts to messages, the `ActorKind` trait describes +/// how messages get into and out of that actor. The three kinds provided by this crate are +/// [`SinkActor`], [`StreamActor`], and [`JointActor`], but these can be extended into other +/// message-passing patterns (e.g. back pressure, single-consumer clients, etc.) without changing +/// the [`Scheduler`]. +/// +/// A kind is constructed when the actor is launched where it does three jobs: +/// 1. Builds the [`Client`](ActorKind::Client) that [`ActorBuilder::launch`] returns. +/// 2. Attaches whatever inbound streams that client needs. A [`SinkActor`], for example, +/// attaches the receiving half of the [`mpsc-style`](tokio::sync::mpsc) channel that backs its +/// [`SinkClient`]. +/// 3. Keeps whatever state is needed to send messages back out. A [`StreamActor`], for example, +/// holds the [`broadcast`](tokio::sync::broadcast) sender that its [`StreamClient`]s listen to. +/// +/// That third piece is stored in the [`Scheduler`], which [`Deref`](std::ops::Deref)s to the kind. +/// Any inherent method a kind defines is therefore callable straight off the `scheduler` that an +/// [`ActorState`] is handed, which is how [`StreamActor::broadcast`] is reached. +pub trait ActorKind: Sized + Sendable { + /// The client that is returned to the caller of [`ActorBuilder::launch`]. + type Client; + + /// The data that this needs in order to be constructed. Kinds that need no config should use + /// `()`, so [`ActorBuilder::config`] does not need to be called. Otherwise, a value of this + /// type must be given to the builder via [`ActorBuilder::config`] before launching. + type Config; -/// A marker type used in the [`ActorState`]. It communicates that the actor should exist for a -/// non-infinite amount of time. The [`Scheduler`] will provide the actor state a method to -/// shutdown. Also, the [`Tracker`](crate::sink::permanent::Tracker)s for request-response style messages will not implictly unwrap -/// responses from their oneshot channels. -#[derive(Debug)] -pub struct Transient; + /// Uses the given config to construct `Self`, its client, and a callback that finishes wiring + /// up the [`Scheduler`]. + /// + /// The kind is needed to build the scheduler, so anything that must be registered *on* the + /// scheduler (most notably inbound streams, which are attached via + /// [`Scheduler::attach_stream`]) can not be done here. Instead, that work is done by the + /// returned closure once the scheduler exists. + fn construct( + config: Self::Config, + ) -> (Self, Self::Client, impl 'static + FnOnce(&mut Scheduler)); +} /// Holds a type that implements [`ActorState`], helps aggregate all data that the actor needs, and -/// then launches the async actor process. When the actor process is launched, a client is returned -/// to the caller. This client's type depends on the actor's type. +/// then spawns the actor in an async task. After spawning the task, a client is returned +/// to the caller. That client's type is the [`Client`](ActorKind::Client) of the state's +/// [`ActorKind`]. #[allow(missing_debug_implementations)] -pub struct ActorBuilder { - /// The type of actor that is being built. This is the same as `A::ActorType` but - /// specialization is not yet supported. - ty: PhantomData, - send: UnboundedSender, - edges: SendableAnyMap, - #[cfg(not(target_family = "wasm"))] - #[allow(clippy::type_complexity)] - broadcast: Option<(broadcast::Sender, broadcast::Receiver)>, - #[cfg(target_family = "wasm")] - #[allow(clippy::type_complexity)] - broadcast: Option<( - broadcast::Sender>, - broadcast::Receiver>, - )>, +pub struct ActorBuilder { recv: Vec>, + config: C, state: A, } /* --------- All actors --------- */ -impl ActorBuilder -where - A: ActorState, -{ +impl ActorBuilder { /// Constructs a new builder for an actor that uses the given state. pub fn new(state: A) -> Self { - let (send, recv) = unbounded_channel(); - let recv = vec![recv.into()]; Self { state, - send, - recv, - broadcast: None, - ty: PhantomData, - edges: SendableAnyMap::new(), + recv: vec![], + config: (), } } +} +impl ActorBuilder { /// Attaches a stream that will be used by the actor once its spawned. No messages will be /// processed until after the actor is launched. - pub fn attach_stream(&mut self, stream: S) + pub fn attach_stream(mut self, stream: S) -> Self where S: SendableFusedStream, I: Into, { - self.recv - .push(ActorStream::Secondary(Box::new(stream.map(|m| m.into())))); + self.recv.push(Box::new(stream.map(|m| m.into()))); + self } - /// Adds a client to the builder, which the state can access later. - pub fn add_edge(&mut self, client: SinkClient) { - _ = self.edges.insert(client); - } - - /// Adds an arbitrary data to the builder, which the state can access later. This method is - /// intended to be used with containers hold that multiple clients of the same type. - /// - /// For example, you can attach a series of actor clients that are indexed using a hashmap. - pub fn add_multi_edge(&mut self, container: C) { - _ = self.edges.insert(container); - } -} - -/* --------- Sink actors --------- */ -impl ActorBuilder -where - A: ActorState, -{ - /// Returns a client for the actor that will be spawned. While the returned client will be able - /// to send messages, those messages will not be processed until after the actor is launched by - /// the builder. - pub fn client(&self) -> SinkClient { - SinkClient::new(self.send.clone()) - } - - /// Launches an actor that uses the given state and returns a client to the actor. - pub fn launch(self) -> SinkClient { + /// Provides the config that the state's [`ActorKind`] needs in order to be constructed at + /// launch. Kinds whose [`Config`](ActorKind::Config) is a unit do not need this method to be + /// called. Calling this method more than once drops the previous config. + pub fn config( + self, + config: >::Config, + ) -> ActorBuilder>::Config> { let Self { - send, recv, state, - edges, - .. + config: _, } = self; - let mut runner = ActorRunner::new(state, edges); - recv.into_iter().for_each(|r| runner.add_stream(r)); - runner.launch(); - SinkClient::new(send) - } -} - -/* --------- Stream actors --------- */ -impl ActorBuilder -where - A: ActorState, -{ - /// Returns a client for the actor that will be spawned. The client will not yield any messages - /// until after the actor is launched and has sent a message. - pub fn client(&mut self) -> StreamClient { - let (_, broad) = self - .broadcast - .get_or_insert_with(|| broadcast::channel(100)); - StreamClient::new(broad.resubscribe()) - } - - /// Launches an actor that uses the given state. Returns a client to the actor. - pub fn launch(self, stream: S) -> StreamClient - where - S: SendableFusedStream, - { - let Self { - mut recv, + ActorBuilder { + recv, + config, state, - broadcast, - edges, - .. - } = self; - let (broad, sub) = broadcast.unwrap_or_else(|| broadcast::channel(100)); - recv.push(ActorStream::Secondary(Box::new(stream))); - let mut runner = ActorRunner::new(state, edges); - runner.add_broadcaster(broad); - recv.into_iter().for_each(|r| runner.add_stream(r)); - runner.launch(); - StreamClient::new(sub) + } } } -/* --------- Joint actors --------- */ -impl ActorBuilder +impl ActorBuilder where - A: ActorState, + K: ActorKind, + A: ActorState, { - /// Returns a stream client for the actor that will be spawned. The client will not yield any - /// messages until after the actor is launched and has sent a message. - pub fn stream_client(&self) -> StreamClient { - StreamClient::new(self.broadcast.as_ref().unwrap().1.resubscribe()) - } - - /// Returns a sink client for the actor that will be spawned. While the returned client will be - /// able to send messages, those messages will not be processed until after the actor is - /// launched by the builder. - pub fn sink_client(&self) -> SinkClient { - SinkClient::new(self.send.clone()) - } - - /// Returns a joint client for the actor that will be spawned. While the returned client will be - /// able to send messages, those messages will not be processed until after the actor is - /// launched by the builder. The client will also not yield any messages until after the actor - /// is launched and has sent a message. - pub fn client(&self) -> SinkClient { - SinkClient::new(self.send.clone()) - } - - /// Launches an actor that uses the given state and stream. Returns a client to the actor. - pub fn launch_with_stream( - mut self, - stream: S, - ) -> JointClient - where - S: SendableFusedStream, - { - self.attach_stream(stream); - self.launch() - } - - /// Launches an actor that uses the given state. Returns a client to the actor. - pub fn launch(self) -> JointClient { + /// Spawns the actor in a separate async task and returns the client of the actor's + /// [`ActorKind`], through which messages can be sent and/or received. + pub fn spawn(self) -> K::Client { let Self { - send, recv, + config, state, - broadcast, - edges, - .. } = self; - let (broad, sub) = broadcast.unwrap_or_else(|| broadcast::channel(100)); - let mut runner = ActorRunner::new(state, edges); - recv.into_iter().for_each(|r| runner.add_stream(r)); - runner.add_broadcaster(broad); - runner.launch(); - let sink = SinkClient::new(send); - let stream = StreamClient::new(sub); - JointClient::new(sink, stream) + let (kind, client, init) = A::ActorKind::construct(config); + let mut runner = ActorRunner::new(state, kind); + init(&mut runner.scheduler); + recv.into_iter().for_each(|r| runner.attach_stream(r)); + runner.spawn(); + client } } diff --git a/src/prelude.rs b/src/prelude.rs index 9b06015..5ac979b 100644 --- a/src/prelude.rs +++ b/src/prelude.rs @@ -1,11 +1,13 @@ //! Re-exports of commonly used items. -pub use crate::{ - compat::SendableFuture, - joint::{JointActor, JointClient}, - oneshot_channel, - scheduler::Scheduler, - sink::{SinkActor, SinkClient}, - stream::{StreamActor, StreamClient}, - ActorBuilder, ActorState, OneshotReceiver, OneshotSender, Permanent, Transient, -}; +pub use crate::compat::SendableFuture; +pub use crate::joint::JointActor; +pub use crate::joint::JointClient; +pub use crate::scheduler::Scheduler; +pub use crate::sink::SinkActor; +pub use crate::sink::SinkClient; +pub use crate::stream::StreamActor; +pub use crate::stream::StreamClient; +pub use crate::ActorBuilder; +pub use crate::ActorKind; +pub use crate::ActorState; diff --git a/src/scheduler.rs b/src/scheduler.rs index 4996b81..ed06423 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -1,42 +1,46 @@ -#[cfg(target_family = "wasm")] -use send_wrapper::SendWrapper; - -use futures::{ - stream::{select_all, Fuse, FusedStream, FuturesUnordered, SelectAll}, - FutureExt, Stream, StreamExt, -}; +use std::future::Future; +use std::ops::Deref; +use std::ops::DerefMut; +use std::pin::Pin; +use std::task::Context; +use std::task::Poll; + +use futures::stream::select_all; +use futures::stream::FuturesUnordered; +use futures::stream::SelectAll; +use futures::FutureExt; +use futures::StreamExt; use instant::Instant; use pin_project::pin_project; -use tokio::sync::{broadcast, mpsc::UnboundedReceiver}; -use tokio_stream::wrappers::UnboundedReceiverStream; - -use std::{ - future::Future, - pin::Pin, - task::{Context, Poll}, -}; - -use crate::{ - compat::{ - sleep_until, spawn_task, Sendable, SendableAnyMap, SendableFusedStream, SendableFuture, - SendableStream, Sleep, - }, - sink::SinkClient, - ActorState, Transient, -}; + +use crate::compat::sleep_until; +use crate::compat::spawn_task; +use crate::compat::Sendable; +use crate::compat::SendableFusedStream; +use crate::compat::SendableFuture; +use crate::compat::Sleep; +use crate::ActorState; type FuturesCollection = FuturesUnordered>>>; -/// Encapulates the different states a scheduler can be. Largely used to communicate how a state -/// wishes to shutdown. -enum SchedulerStatus { - Alive, - Marked, - MarkedToFinish, -} +#[allow(type_alias_bounds)] +pub(crate) type ActorStream = Box>; -/// The primary bookkeeper for the actor. The state attach stream and queue manage futures that -/// will be managed by the Scheduler. +/// The primary bookkeeper for the actor. Everything that can produce a message for the actor lives +/// here: the streams that have been attached to it and the futures that it has queued. An +/// [`ActorState`](crate::ActorState) is handed a `&mut Scheduler` in each of its methods and uses +/// it to attach more streams ([`attach_stream`](Scheduler::attach_stream)), queue more work +/// ([`queue_task`](Scheduler::queue_task), [`manage_future`](Scheduler::manage_future), +/// [`schedule`](Scheduler::schedule)), and ask to be shut down +/// ([`shutdown`](Scheduler::shutdown), [`shutdown_and_finish`](Scheduler::shutdown_and_finish)). +/// +/// The scheduler also holds the actor's [`ActorKind`](crate::ActorKind) and [`Deref`]s to it. This +/// is how a state reaches whatever its kind offers for talking back to clients: for a +/// [`StreamActor`](crate::stream::StreamActor) or [`JointActor`](crate::joint::JointActor), that +/// is `broadcast`, so `scheduler.broadcast(msg)` sends a message to every listening client. A +/// [`SinkActor`](crate::sink::SinkActor) has no such methods, so a sink actor's scheduler simply +/// offers nothing extra. +/// /// The scheduler also tracks if it is possible that no other message will be /// yielded for the actor to process. If it finds itself in a state where all streams are closed /// and there are no queued futures, it will close the actor; otherwise, the deadlocked actor will @@ -47,16 +51,11 @@ enum SchedulerStatus { #[allow(missing_debug_implementations)] pub struct Scheduler { /// The inbound streams to the actor. - recv: SelectAll>>, + recv: SelectAll>, /// Futures that the actor has queued that will yield a message. queue: FuturesCollection, /// Futures that yield nothing that the scheduler will manage and poll for the actor. tasks: FuturesCollection<()>, - /// The manager for outbound messages that will be broadcast from the actor. - outbound: Option>, - /// Stores edges in the form of `EdgeType`s. This is used to access connections to other actors - /// at runtime without needing to embed them into the actor state directly. - edges: SendableAnyMap, /// The number of stream that could yield a message for the actor to process. Once this and the /// `future_count` hit both reach zero, the actor is dead as it can no longer process any /// messages. @@ -67,66 +66,49 @@ pub struct Scheduler { future_count: usize, /// Tracks the status of the scheduler, mostly used to track how the state wants to shutdown. status: SchedulerStatus, + /// The actor's kind. This holds whatever state the kind needs to communicate with its clients + /// (such as a broadcast sender) and is exposed to the actor state via `Deref`/`DerefMut`. + actor_kind: A::ActorKind, } -struct OutboundQueue { - #[cfg(not(target_family = "wasm"))] - send: broadcast::Sender, - #[cfg(target_family = "wasm")] - send: broadcast::Sender>, -} - -impl OutboundQueue { - #[cfg(not(target_family = "wasm"))] - fn new(send: broadcast::Sender) -> Self { - Self { send } - } - - #[cfg(target_family = "wasm")] - fn new(send: broadcast::Sender>) -> Self { - Self { send } - } - - fn send(&mut self, msg: M) { - #[cfg(target_family = "wasm")] - let msg = SendWrapper::new(msg); - let _ = self.send.send(msg); - } +/// Encapulates the different states a scheduler can be. Largely used to communicate how a state +/// wishes to shutdown. +enum SchedulerStatus { + Alive, + Marked, + MarkedToFinish, } /// The container for the actor state and its scheduler. The runner polls the scheduler, aids in /// bookkeeping if the actor is dead or not, and passes messages off to the state. pub(crate) struct ActorRunner { state: A, - scheduler: Scheduler, + pub(super) scheduler: Scheduler, } impl ActorRunner { - pub(crate) fn new(state: A, edges: SendableAnyMap) -> Self { - let scheduler = Scheduler::new(edges); + pub(crate) fn new(state: A, kind: A::ActorKind) -> Self { + let scheduler = Scheduler::new(kind); Self { scheduler, state } } - #[cfg(not(target_family = "wasm"))] - pub(crate) fn add_broadcaster(&mut self, broad: broadcast::Sender) { - self.scheduler.outbound = Some(OutboundQueue::new(broad)); - } - - #[cfg(target_family = "wasm")] - pub(crate) fn add_broadcaster(&mut self, broad: broadcast::Sender>) { - self.scheduler.outbound = Some(OutboundQueue::new(broad)); - } - - pub(crate) fn add_stream(&mut self, stream: ActorStream) { + pub(crate) fn attach_stream(&mut self, stream: ActorStream) { self.scheduler.attach_stream_inner(stream); } - pub(crate) fn launch(self) { + pub(crate) fn spawn(self) { spawn_task(self.run()) } async fn run(mut self) { self.state.start_up(&mut self.scheduler).await; + // It is possible that nothing was originally scheduled in the scheduler or that the state + // consumed all messages during start up. Either way, the scheduler is empty and need the + // actor needs to be closed. + if self.scheduler.is_dead() { + self.close().await; + return + } loop { match self.scheduler.next().await { Some(msg) => self.state.process(&mut self.scheduler, msg).await, @@ -149,16 +131,16 @@ impl ActorRunner { impl Scheduler { /// The constructor for the scheduler. - fn new(edges: SendableAnyMap) -> Self { + fn new(actor_kind: A::ActorKind) -> Self { let recv = select_all([]); let queue = FuturesCollection::new(); let tasks = FuturesCollection::new(); + Self { recv, queue, tasks, - edges, - outbound: None, + actor_kind, stream_count: 0, future_count: 0, status: SchedulerStatus::Alive, @@ -185,7 +167,7 @@ impl Scheduler { } /// Yields the next message to be processed by the actor state. - async fn next(&mut self) -> Option { + pub async fn next(&mut self) -> Option { loop { if self.is_dead() { return None; @@ -220,9 +202,9 @@ impl Scheduler { /// between the queued futures and attached streams. The first to yield an item is the first to /// be processed. For this reason, the futures queued this way must be `'static`, i.e. they /// can't reference the actor's state. - pub fn queue_task(&mut self, fut: F) + pub fn await_message(&mut self, fut: F) where - F: Sendable + Future, + F: SendableFuture, I: 'static + Into, { self.future_count += 1; @@ -247,23 +229,23 @@ impl Scheduler { /// Attaches a stream that will be polled and managed by the scheduler. Messages yielded by the /// streams must be able to be converted into the actor's message type so that the actor can - /// process it. The given stream must be a [`FusedStream`]; however, the scheduler requires a + /// process it. The given stream must be a [`FusedStream`](futures::stream::FusedStream); however, the scheduler requires a /// stronger invariant than that given by `FusedStream`. The scheduler will mark a stream as /// "done" once the stream yields its first `None`. After that, the scheduler will never poll /// that stream again. pub fn attach_stream(&mut self, stream: S) where - S: SendableStream + FusedStream, + S: SendableFusedStream, I: Into, { - let stream = ActorStream::Secondary(Box::new(stream.map(|m| m.into()))); + let stream = Box::new(stream.map(|m| m.into())); self.attach_stream_inner(stream) } /// Adds an actor stream to the scheduler. fn attach_stream_inner(&mut self, stream: ActorStream) { self.stream_count += 1; - self.recv.push(stream.fuse()); + self.recv.push(stream); } /// Schedules a message to be given to the actor to process at a given time. @@ -275,79 +257,6 @@ impl Scheduler { self.queue.push(Box::pin(Timer::new(deadline, msg.into()))); } - /// Broadcasts a message to all listening clients. If the message fails to send, the message - /// will be dropped. - /// - /// Note: This method does nothing if the actor is a [`SinkActor`](crate::sink::SinkActor). - /// [`StreamActor`](crate::stream::StreamActor)s and [`JointActor`](crate::joint::JointActor) - /// will be able to broadcast. - pub fn broadcast(&mut self, msg: M) - where - M: Into, - { - if let Some(out) = self.outbound.as_mut() { - out.send(msg.into()) - } - } - - /// Adds a client to the set of connections to other actors, which the state can access later. - pub fn add_edge(&mut self, client: SinkClient) { - _ = self.edges.insert(client); - } - - /// Gets a reference to a sink client to another actor. - pub fn get_edge(&self) -> Option<&SinkClient> { - self.edges.get::>() - } - - /// Gets mutable reference to a sink client to another actor. - pub fn get_edge_mut( - &mut self, - ) -> Option<&mut SinkClient> { - self.edges.get_mut::>() - } - - /// Removes a client to the set of connections to other actors. - pub fn remove_edge(&mut self) { - _ = self.edges.remove::>(); - } - - /// Adds an arbitrary data to the set of connections to other actors, which the state can - /// access later. This method is intended to be used with containers hold that multiple clients - /// of the same type. - /// - /// For example, you can attach a series of actor clients that are indexed using a hashmap. - pub fn add_multi_edge(&mut self, container: C) { - _ = self.edges.insert(container); - } - - /// Gets a reference to an arbitary type held in the container that holds connections to other - /// actors. This method is intended to be used with containers that hold multiple clients of - /// the same type. - /// - /// For example, you can store and access a series of actor clients that are indexed using a - /// hashmap. - pub fn get_multi_edge(&self) -> Option<&C> { - self.edges.get::() - } - - /// Gets a mutable reference to an arbitary type held in the container that holds connections to other - /// actors. This method is intended to be used with containers that hold multiple clients of - /// the same type. - pub fn get_multi_edge_mut(&mut self) -> Option<&mut C> { - self.edges.get_mut::() - } - - /// Adds a piece of arbitrary data from the set of connections to other actors. - pub fn remove_multi_edge(&mut self) { - _ = self.edges.remove::(); - } -} - -impl Scheduler -where - A: ActorState, -{ /// Marks the actor as ready to shutdown. After the state finishes processing the current /// message, actor process will shutdown. Any unprocessed messages will be dropped, all /// attached streams will be closed, all futures that will yield a message will be cancelled, @@ -369,25 +278,23 @@ where } } -impl From> for ActorStream { - fn from(value: UnboundedReceiver) -> Self { - Self::Main(UnboundedReceiverStream::new(value)) - } -} +impl Deref for Scheduler +where + A: ActorState, +{ + type Target = A::ActorKind; -pub(crate) enum ActorStream { - Main(UnboundedReceiverStream), - Secondary(Box>), + fn deref(&self) -> &Self::Target { + &self.actor_kind + } } -impl Stream for ActorStream { - type Item = M; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - match *self { - ActorStream::Main(ref mut stream) => Pin::new(stream).poll_next(cx), - ActorStream::Secondary(ref mut stream) => Pin::new(stream).poll_next(cx), - } +impl DerefMut for Scheduler +where + A: ActorState, +{ + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.actor_kind } } diff --git a/src/sink.rs b/src/sink.rs index dede6f0..c43396f 100644 --- a/src/sink.rs +++ b/src/sink.rs @@ -1,23 +1,53 @@ //! Actors that are only sent messages (either fire-and-forget messages or request-response //! messages). +use std::fmt::Debug; +use std::future::Future; +use std::pin::Pin; +use std::task::Context; +use std::task::Poll; -use std::marker::PhantomData; - +use futures::stream::StreamExt; use tokio::sync::mpsc::UnboundedSender; +use tokio::sync::oneshot; +use tokio_stream::wrappers::UnboundedReceiverStream; -use crate::{oneshot_channel, OneshotSender, Permanent, Transient}; +use crate::ActorKind; +use crate::ActorState; +use crate::Scheduler; -/// A marker type used by the [`ActorBuilder`](crate::ActorBuilder) to know what kind of -/// [`ActorState`](crate::ActorState) it is dealing with. A sink actor is one that receives +/// The [`ActorKind`] for actors that only receive messages. A sink actor is one that receives /// messages from other parts of the application. By adding a oneshot channel to the message, /// the actor can respond with a particular piece of data. This allows for type-safe communication /// between different parts of your program. /// /// The client of a [`SinkActor`] is the [`SinkClient`]. This client implements methods that allow /// for the sending of messages to this client. Communication between a sink client and sink actor -/// uses an MPSC-style channel (see [`mpsc::channel`](tokio::sync::mpsc)). +/// uses an MPSC-style channel (see [`mpsc::channel`](tokio::sync::mpsc)); constructing this kind +/// creates that channel, hands the sending half to the client, and attaches the receiving half to +/// the actor's [`Scheduler`]. +/// +/// Unlike the other kinds, a sink actor sends nothing back to its clients outside of the oneshot +/// channels carried by its own messages, so this type holds no state of its own. #[derive(Debug)] -pub struct SinkActor; +pub struct SinkActor {} + +impl ActorKind for SinkActor { + type Client = SinkClient; + type Config = (); + + fn construct( + (): Self::Config, + ) -> (Self, Self::Client, impl 'static + FnOnce(&mut Scheduler)) { + let this = Self {}; + + let (send, recv) = tokio::sync::mpsc::unbounded_channel(); + let client = SinkClient::new(send); + let func = move |scheduler: &mut Scheduler| { + scheduler.attach_stream(UnboundedReceiverStream::new(recv).fuse()); + }; + (this, client, func) + } +} /// A client to an actor. This client sends messages to the actor and supports two styles of /// messaging. The first is fire-and-forget messages. These messages are sent to the client @@ -32,7 +62,6 @@ pub struct SinkActor; /// message type. Say you have an actor like the one below. You can send messages to that actor /// like so: /// ```ignore -/// # extern crate derive_more; /// # use std::collections::HashMap; /// # use troupe::prelude::*; /// # use derive_more::From; @@ -45,19 +74,17 @@ pub struct SinkActor; /// Get(usize, OneshotSender>), /// Delete(usize), /// } -/// # #[async_trait] +/// /// # impl ActorState for CacheState { +/// # type ActorKind = SinkActor; /// # type Message = CacheCommand; -/// # type ActorType = SinkActor; -/// # type Permanence = Permanent; -/// # type Output = (); /// # /// # async fn process(&mut self, scheduler: &mut Scheduler, msg: Self::Message) { () } /// # } +/// // `SinkActor`'s config is `()`, so the builder can be launched directly. +/// let client: SinkClient = ActorBuilder::new(CacheState::default()).launch(); /// -/// let client = ActorBuilder::new(CacheState::default()).launch(); -/// -/// // Sends CacheCommand::Inset(42, "Hello world") +/// // Sends CacheCommand::Insert(42, "Hello world") /// client.send((42, String::from("Hello World"))); /// // Sends CacheCommand::Get(42, OneshotSender) and returns a tracker which will listen for a /// // response from the actor. @@ -66,17 +93,13 @@ pub struct SinkActor; /// client.send(42); /// ``` #[derive(Debug)] -pub struct SinkClient { - ty: PhantomData, +pub struct SinkClient { send: UnboundedSender, } -impl SinkClient { +impl SinkClient { pub(crate) fn new(send: UnboundedSender) -> Self { - Self { - send, - ty: PhantomData, - } + Self { send } } /// Returns if the actor that the client is connected to is dead or not. @@ -89,118 +112,48 @@ impl SinkClient { pub fn send(&self, msg: impl Into) -> bool { self.send.send(msg.into()).is_ok() } -} -impl SinkClient { - /// Sends a request-response style message to a [`Permanent`] actor. The given data is paired - /// with a one-time use channel and sent to the actor. A [`Tracker`](permanent::Tracker) that - /// will receive a response from the actor is returned. - /// - /// Note: Since this client is one for a permanent actor, there is an implicit unwrap once the - /// tracker receives a message from the actor. If the actor drops the other half of the channel - /// or has died somehow (likely from a panic), the returned tracker will panic too. So, it is - /// important that the actor always sends back a message - pub fn track(&self, msg: I) -> permanent::Tracker + /// Sends a request-response style message to an actor. The given data is paired with a + /// one-time use channel and sent to the actor. A [`Tracker`] that will receive a response from + /// the actor is returned. + pub fn track(&self, msg: I) -> Tracker where - M: From<(I, OneshotSender)>, + M: From<(I, oneshot::Sender)>, { - let (send, recv) = oneshot_channel(); + let (send, recv) = oneshot::channel(); let msg = M::from((msg, send)); let _ = self.send(msg); - permanent::Tracker::new(recv) + Tracker::new(recv) } } -impl SinkClient { - /// Sends a request-response style message to a [`Transient`] actor. The given data is paired - /// with a one-time use channel and sent to the actor. A [`Tracker`](transient::Tracker) that - /// will receive a response from the actor is returned. - pub fn track(&self, msg: I) -> transient::Tracker - where - M: From<(I, OneshotSender)>, - { - let (send, recv) = oneshot_channel(); - let msg = M::from((msg, send)); - let _ = self.send(msg); - transient::Tracker::new(recv) - } -} - -impl Clone for SinkClient { +impl Clone for SinkClient { fn clone(&self) -> Self { Self::new(self.send.clone()) } } -/// A module for things used to interact with the [`Permanent`] actors. -pub mod permanent { - use std::{ - future::Future, - pin::Pin, - task::{Context, Poll}, - }; - - use crate::OneshotReceiver; - - /// A tracker for a request-response style message sent to a [`Permanent`](crate::Permanent) actor. - /// - /// Note: This tracker implicitly unwraps the message produced by its channel receiver. If the - /// actor drops the other half of the channel or has died somehow (likely from a panic), this - /// tracker will panic when polled. - #[derive(Debug)] - pub struct Tracker { - recv: OneshotReceiver, - } - - impl Tracker { - /// A constructor for the tracker. - pub(crate) fn new(recv: OneshotReceiver) -> Self { - Self { recv } - } - } - - impl Future for Tracker { - type Output = T; - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - Pin::new(&mut self.recv).poll(cx).map(Result::unwrap) - } - } +/// A tracker for a request-response style message sent to an actor. +/// +/// Note: This tracker might be created after a failed attempt to send a message to a dead +/// actor. This means that the tracker will return `None` when polled; however, that does not +/// mean that the message was successfully received by the actor. +#[derive(Debug)] +pub struct Tracker { + recv: oneshot::Receiver, } -/// A module for things used to interact with the [`Transient`] actors. -pub mod transient { - use std::{ - fmt::Debug, - future::Future, - pin::Pin, - task::{Context, Poll}, - }; - - use crate::OneshotReceiver; - - /// A tracker for a request-response style message sent to a [`Transient`](crate::Transient) actor. - /// - /// Note: This tracker might be created after a failed attempt to send a message to a dead - /// actor. This means that the tracker will return `None` when polled; however, that does not - /// mean that the message was successfully received by the actor. - #[derive(Debug)] - pub struct Tracker { - recv: OneshotReceiver, - } - - impl Tracker { - /// A constuctor for the tracker. - pub(crate) fn new(recv: OneshotReceiver) -> Self { - Self { recv } - } +impl Tracker { + /// A constuctor for the tracker. + pub(crate) fn new(recv: oneshot::Receiver) -> Self { + Self { recv } } +} - impl Future for Tracker { - type Output = Option; +impl Future for Tracker { + type Output = Option; - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - Pin::new(&mut self.recv).poll(cx).map(Result::ok) - } + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + Pin::new(&mut self.recv).poll(cx).map(Result::ok) } } diff --git a/src/stream.rs b/src/stream.rs index 05f97cd..e45d650 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -1,33 +1,87 @@ //! Actors that broadcast messages. -use std::{ - fmt::Debug, - pin::Pin, - task::{Context, Poll}, -}; - -#[cfg(target_family = "wasm")] -use send_wrapper::SendWrapper; - -use futures::{ready, Stream, StreamExt}; +use std::fmt::Debug; +use std::pin::Pin; +use std::task::Context; +use std::task::Poll; + +use futures::ready; +use futures::Stream; +use futures::StreamExt; use tokio::sync::broadcast; use tokio_stream::wrappers::errors::BroadcastStreamRecvError; use crate::compat::Sendable; +use crate::ActorKind; +use crate::ActorState; +use crate::Scheduler; -/// A marker type used by the [`ActorBuilder`](crate::ActorBuilder) to know what kind of -/// [`ActorState`](crate::ActorState) it is dealing with. A stream actor is one that receives -/// messages from one or more streams and then forwards messages to its clients. +#[cfg(not(target_family = "wasm"))] +pub(crate) type Broadcastee = M; +#[cfg(target_family = "wasm")] +pub(crate) type Broadcastee = send_wrapper::SendWrapper; + +/// The [`ActorKind`] for actors that only broadcast messages. A stream actor is one that receives +/// messages from one or more streams and then forwards messages of type `M` to its clients. /// /// The client of a [`StreamActor`] is the [`StreamClient`]. This client implements methods for /// receiving methods that are "forwarded" by the actor. Unlike the /// [`SinkActor`](crate::sink::SinkActor), stream actors and clients don't directly support /// request/response style communication. Communication between a stream actor and client(s) can be /// modelled with a broadcast-style channel (see [`broadcast::channel`]). +/// +/// This kind holds the sending half of that broadcast channel. Because it is stored in the +/// [`Scheduler`], which derefs to it, an [`ActorState`] broadcasts by calling +/// [`broadcast`](StreamActor::broadcast) on the scheduler it is given. +/// +/// A stream actor attaches no streams of its own. Every message it processes comes from a stream +/// given to [`ActorBuilder::attach_stream`](crate::ActorBuilder::attach_stream) or +/// [`Scheduler::attach_stream`], or from a future queued in the scheduler. Once all of those run +/// dry, the actor is closed. #[derive(Debug)] -pub struct StreamActor; +pub struct StreamActor { + pub(crate) broadcast: broadcast::Sender>, +} + +impl ActorKind for StreamActor { + type Client = StreamClient; + type Config = (); + + fn construct( + (): Self::Config, + ) -> (Self, Self::Client, impl 'static + FnOnce(&mut Scheduler)) { + let (send, recv) = broadcast::channel(10); + + let client = StreamClient::new(recv); + let this = Self { broadcast: send }; + let init = move |_scheduler: &mut Scheduler| {}; + + (this, client, init) + } +} + +impl StreamActor { + /// Broadcasts a message to all listening clients. If the message fails to send, the message + /// will be dropped. + /// + /// This is normally reached through the [`Scheduler`], which derefs to this type, rather than + /// on a `StreamActor` directly. + pub fn broadcast(&mut self, msg: impl Into) { + #[cfg(not(target_family = "wasm"))] + let _ = self.broadcast.send(msg.into()); + #[cfg(target_family = "wasm")] + let _ = self + .broadcast + .send(send_wrapper::SendWrapper::new(msg.into())); + } +} /// A client that receives messages from an actor that broadcasts them. +/// +/// This client implements [`Stream`], yielding `Result`. The `Err` case reports that the +/// client lagged behind the actor and that the contained number of messages were missed, not that +/// the actor failed. The stream ends once the actor has closed and its buffered messages have been +/// drained. #[derive(Debug)] pub struct StreamClient { recv: BroadcastStream, @@ -38,30 +92,16 @@ pub struct StreamClient { /// receiver. struct BroadcastStream { /// A copy of the original channel, used for cloning the client. - #[cfg(not(target_family = "wasm"))] - copy: broadcast::Receiver, - #[cfg(target_family = "wasm")] - copy: broadcast::Receiver>, - #[cfg(not(target_family = "wasm"))] + copy: broadcast::Receiver>, /// The stream that is polled. - inner: tokio_stream::wrappers::BroadcastStream, - #[cfg(target_family = "wasm")] - inner: tokio_stream::wrappers::BroadcastStream>, + inner: tokio_stream::wrappers::BroadcastStream>, } impl StreamClient where M: Sendable + Clone, { - #[cfg(not(target_family = "wasm"))] - pub(crate) fn new(recv: broadcast::Receiver) -> Self { - Self { - recv: BroadcastStream::new(recv), - } - } - - #[cfg(target_family = "wasm")] - pub(crate) fn new(recv: broadcast::Receiver>) -> Self { + pub(crate) fn new(recv: broadcast::Receiver>) -> Self { Self { recv: BroadcastStream::new(recv), } @@ -72,15 +112,7 @@ impl BroadcastStream where M: Sendable + Clone, { - #[cfg(not(target_family = "wasm"))] - fn new(stream: broadcast::Receiver) -> Self { - let copy = stream.resubscribe(); - let inner = tokio_stream::wrappers::BroadcastStream::new(stream); - Self { copy, inner } - } - - #[cfg(target_family = "wasm")] - fn new(stream: broadcast::Receiver>) -> Self { + fn new(stream: broadcast::Receiver>) -> Self { let copy = stream.resubscribe(); let inner = tokio_stream::wrappers::BroadcastStream::new(stream); Self { copy, inner } diff --git a/tests/joint.rs b/tests/joint.rs index 6dd47bb..67d8366 100644 --- a/tests/joint.rs +++ b/tests/joint.rs @@ -1,7 +1,8 @@ use futures::StreamExt; use instant::Duration; use tokio::sync::oneshot::error::TryRecvError; -use troupe::{compat::sleep_for, prelude::*}; +use troupe::compat::sleep_for; +use troupe::prelude::*; #[derive(Debug, PartialEq, Eq)] struct Started; @@ -18,28 +19,19 @@ struct DummyJoint { } impl ActorState for DummyJoint { - type ActorType = JointActor; - type Permanence = Permanent; + type ActorKind = JointActor; type Message = Processed; - type Output = Processed; - fn start_up(&mut self, _: &mut Scheduler) -> impl SendableFuture { + async fn start_up(&mut self, _: &mut Scheduler) { self.started.take().unwrap().send(Started).unwrap(); - std::future::ready(()) } - fn process( - &mut self, - scheduler: &mut Scheduler, - msg: Self::Message, - ) -> impl SendableFuture { + async fn process(&mut self, scheduler: &mut Scheduler, msg: Self::Message) { scheduler.broadcast(msg); - std::future::ready(()) } - fn finalize(self, _: &mut Scheduler) -> impl SendableFuture { + async fn finalize(self, _: &mut Scheduler) { self.completed.send(Completed).unwrap(); - std::future::ready(()) } } @@ -52,8 +44,7 @@ fn are_send() { fn is_send() {} is_send::(); - is_send::(); - is_send::(); + is_send::>(); is_send::(); is_send::(); is_send::(); @@ -70,7 +61,7 @@ async fn startup_and_teardown() { assert_eq!(Err(TryRecvError::Empty), started_recv.try_recv()); assert_eq!(Err(TryRecvError::Empty), comped_recv.try_recv()); let stream = futures::stream::iter(std::iter::once(Processed)).fuse(); - let mut client = ActorBuilder::new(state).launch_with_stream(stream); + let mut client = ActorBuilder::new(state).attach_stream(stream).launch(); /* ----- Successful startup test ----- */ tokio::select! { _ = sleep() => { diff --git a/tests/sink.rs b/tests/sink.rs index 83737cb..33294f4 100644 --- a/tests/sink.rs +++ b/tests/sink.rs @@ -1,9 +1,7 @@ use instant::Duration; use tokio::sync::oneshot::error::TryRecvError; -use troupe::{ - compat::{sleep_for, SendableFuture}, - prelude::*, -}; +use troupe::compat::sleep_for; +use troupe::prelude::*; #[derive(Debug, PartialEq, Eq)] struct Started; @@ -17,30 +15,17 @@ struct DummySink { } impl ActorState for DummySink { - type ActorType = SinkActor; - type Permanence = Permanent; + type ActorKind = SinkActor; type Message = (); - type Output = (); - fn start_up( - &mut self, - _: &mut Scheduler, - ) -> impl troupe::compat::SendableFuture { + async fn start_up(&mut self, _: &mut Scheduler) { self.started.take().unwrap().send(Started).unwrap(); - std::future::ready(()) } - fn process( - &mut self, - _: &mut Scheduler, - _: Self::Message, - ) -> impl SendableFuture { - std::future::ready(()) - } + async fn process(&mut self, _: &mut Scheduler, _: Self::Message) {} - fn finalize(self, _: &mut Scheduler) -> impl SendableFuture { + async fn finalize(self, _: &mut Scheduler) { self.completed.send(Completed).unwrap(); - std::future::ready(()) } } diff --git a/tests/stream.rs b/tests/stream.rs index 0fb04ab..905f937 100644 --- a/tests/stream.rs +++ b/tests/stream.rs @@ -1,7 +1,9 @@ use futures::StreamExt; -use instant::{Duration, Instant}; +use instant::Duration; +use instant::Instant; use tokio::sync::oneshot::error::TryRecvError; -use troupe::{compat::sleep_for, prelude::*}; +use troupe::compat::sleep_for; +use troupe::prelude::*; #[derive(Debug, PartialEq, Eq)] struct Started; @@ -18,10 +20,8 @@ struct DummyStream { } impl ActorState for DummyStream { - type ActorType = StreamActor; - type Permanence = Permanent; + type ActorKind = StreamActor; type Message = Processed; - type Output = Processed; async fn start_up(&mut self, _: &mut Scheduler) { self.started.take().unwrap().send(Started).unwrap(); @@ -51,7 +51,7 @@ async fn startup_and_teardown() { assert_eq!(Err(TryRecvError::Empty), started_recv.try_recv()); assert_eq!(Err(TryRecvError::Empty), comped_recv.try_recv()); let stream = futures::stream::iter(std::iter::once(Processed)).fuse(); - let mut client = ActorBuilder::new(state).launch(stream); + let mut client = ActorBuilder::new(state).attach_stream(stream).launch(); /* ----- Successful startup test ----- */ tokio::select! { _ = sleep() => { @@ -92,22 +92,15 @@ struct Counter { } impl ActorState for Counter { - type ActorType = StreamActor; - type Permanence = Transient; + type ActorKind = StreamActor; type Message = (); - type Output = usize; - fn start_up(&mut self, scheduler: &mut Scheduler) -> impl SendableFuture { + async fn start_up(&mut self, scheduler: &mut Scheduler) { self.started.take().unwrap().send(Started).unwrap(); scheduler.schedule(Instant::now() + Duration::from_millis(5), ()); - std::future::ready(()) } - fn process( - &mut self, - scheduler: &mut Scheduler, - (): Self::Message, - ) -> impl SendableFuture { + async fn process(&mut self, scheduler: &mut Scheduler, (): Self::Message) { println!("Processing message!!"); let next = self.count + 1; scheduler.broadcast(std::mem::replace(&mut self.count, next)); @@ -115,12 +108,10 @@ impl ActorState for Counter { if self.count > 10 { scheduler.shutdown(); } - std::future::ready(()) } - fn finalize(self, _: &mut Scheduler) -> impl SendableFuture { + async fn finalize(self, _: &mut Scheduler) { self.completed.send(Completed).unwrap(); - std::future::ready(()) } } @@ -135,10 +126,8 @@ async fn stop_after_ten_messages() { }; assert_eq!(Err(TryRecvError::Empty), started_recv.try_recv()); assert_eq!(Err(TryRecvError::Empty), comped_recv.try_recv()); - let stream = futures::stream::iter(std::iter::empty()).fuse(); - let mut builder = ActorBuilder::new(state); - let _client = builder.client(); - let mut client = builder.launch(stream); + let stream = futures::stream::iter(std::iter::empty::<()>()).fuse(); + let mut client = ActorBuilder::new(state).attach_stream(stream).launch(); /* ----- Successful startup test ----- */ tokio::select! { _ = sleep() => { diff --git a/tmp b/tmp new file mode 100644 index 0000000..85728cd --- /dev/null +++ b/tmp @@ -0,0 +1,6 @@ + +// TODO: +// - Make ActorState::start_up return V +// - Normalize the language between docs and across methods, e.g. "ActorBuilder::launch" conflicts +// with "spawning an actor" +// - Make Unbound/backpressure clients