From aefb6e1a4bc61586a5cfb45e98536a38b48208d7 Mon Sep 17 00:00:00 2001 From: "crashkeys.dev" Date: Thu, 12 Mar 2026 21:09:51 +0100 Subject: [PATCH] repo: formatting changes. I set the formatter to use hard tabs. This is so my editor can display them with a width of two without necessarily breaking this for anyone else who wants to work with my code. The indentation could get so deep that the code had begun being very hard to read on my laptop. No more! --- rustfmt.toml | 1 + src/data.rs | 18 +- src/lib.rs | 150 +++++----- src/main.rs | 2 +- src/sink.rs | 264 ++++++++--------- src/storage.rs | 782 ++++++++++++++++++++++++------------------------- 6 files changed, 609 insertions(+), 608 deletions(-) create mode 100644 rustfmt.toml diff --git a/rustfmt.toml b/rustfmt.toml new file mode 100644 index 0000000..218e203 --- /dev/null +++ b/rustfmt.toml @@ -0,0 +1 @@ +hard_tabs = true diff --git a/src/data.rs b/src/data.rs index 15c9bb8..e4d3dcf 100644 --- a/src/data.rs +++ b/src/data.rs @@ -6,19 +6,19 @@ pub struct Quote(String); impl> From for Quote { - fn from(value: S) -> Self { - Self(value.as_ref().to_owned()) - } + fn from(value: S) -> Self { + Self(value.as_ref().to_owned()) + } } impl From for String { - fn from(value: Quote) -> Self { - value.0 - } + fn from(value: Quote) -> Self { + value.0 + } } impl Quote { - pub fn get(&self) -> &str { - &self.0 - } + pub fn get(&self) -> &str { + &self.0 + } } diff --git a/src/lib.rs b/src/lib.rs index 89931de..ececcca 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,92 +3,92 @@ pub mod sink; pub mod storage; pub mod run { - use std::time::Duration; + use std::time::Duration; - use crate::sink::{BskySink, PostQuote, SinkManager, StdoutSink}; - use crate::storage::{ - FetchQuote, QuoteCycle, queue::MemoryQueueStorage, source::FsFilterSourceManager, - }; - use cron_lite::CronEvent; - use futures::StreamExt; - use kameo::prelude::*; - use tokio::time::timeout; + use crate::sink::{BskySink, PostQuote, SinkManager, StdoutSink}; + use crate::storage::{ + FetchQuote, QuoteCycle, queue::MemoryQueueStorage, source::FsFilterSourceManager, + }; + use cron_lite::CronEvent; + use futures::StreamExt; + use kameo::prelude::*; + use tokio::time::timeout; - pub async fn entrypoint() -> Result<(), Box> { - // TODO: Clean up this function's internals. - // The current structure is alright, but it was stitched together - // quickly just to confirm that everything is functioning as it should. - let use_bsky = std::env::var("USE_BLUESKY").unwrap_or("0".to_string()) == "1"; - let bsky = if use_bsky { - Some(BskySink::spawn( - BskySink::new_session( - std::env::var("BLUESKY_USERNAME").expect("Bluesky username not supplied"), - std::env::var("BLUESKY_PASSWORD") - .expect("Bluesky application password not supplied"), - ) - .await - .expect("Could not connect to Bluesky with supplied credentials"), - )) - } else { - None - }; + pub async fn entrypoint() -> Result<(), Box> { + // TODO: Clean up this function's internals. + // The current structure is alright, but it was stitched together + // quickly just to confirm that everything is functioning as it should. + let use_bsky = std::env::var("USE_BLUESKY").unwrap_or("0".to_string()) == "1"; + let bsky = if use_bsky { + Some(BskySink::spawn( + BskySink::new_session( + std::env::var("BLUESKY_USERNAME").expect("Bluesky username not supplied"), + std::env::var("BLUESKY_PASSWORD") + .expect("Bluesky application password not supplied"), + ) + .await + .expect("Could not connect to Bluesky with supplied credentials"), + )) + } else { + None + }; - let sink = { - let stdout = StdoutSink::spawn(StdoutSink); - SinkManager::spawn(SinkManager::new(Some(stdout), bsky)) - }; + let sink = { + let stdout = StdoutSink::spawn(StdoutSink); + SinkManager::spawn(SinkManager::new(Some(stdout), bsky)) + }; - let cycle = { - let source = FsFilterSourceManager::spawn(FsFilterSourceManager::default()); - let queue = MemoryQueueStorage::spawn(MemoryQueueStorage::new()); + let cycle = { + let source = FsFilterSourceManager::spawn(FsFilterSourceManager::default()); + let queue = MemoryQueueStorage::spawn(MemoryQueueStorage::new()); - QuoteCycle::spawn(QuoteCycle::with_thread_rng(source, queue)) - }; + QuoteCycle::spawn(QuoteCycle::with_thread_rng(source, queue)) + }; - use cron_lite::Schedule; - const POSTING_TIMEOUT: Duration = Duration::from_secs(60); - const POSTING_INTERVAL: &str = "*/10 * * * * * *"; - let schedule = - Schedule::new(POSTING_INTERVAL).expect("Schedule should be a valid cron expression"); - let now = chrono::Utc::now(); + use cron_lite::Schedule; + const POSTING_TIMEOUT: Duration = Duration::from_secs(60); + const POSTING_INTERVAL: &str = "*/10 * * * * * *"; + let schedule = + Schedule::new(POSTING_INTERVAL).expect("Schedule should be a valid cron expression"); + let now = chrono::Utc::now(); - let mut tick_stream = schedule.stream(&now); + let mut tick_stream = schedule.stream(&now); - while let Some(tick) = tick_stream.next().await { - if let CronEvent::Missed(missed_at) = tick { - eprintln!( - "Missed event tick at {}. Current time: {}. Skipping post.", - missed_at, - chrono::Utc::now() - ); - continue; - } + while let Some(tick) = tick_stream.next().await { + if let CronEvent::Missed(missed_at) = tick { + eprintln!( + "Missed event tick at {}. Current time: {}. Skipping post.", + missed_at, + chrono::Utc::now() + ); + continue; + } - // We store the code to perform the next posting iteration as one atomic future which we wrap with a timeout. - // This means that, if we miss a posting window due to the timeout, we will not get multiple consecutive or late posts. - let next_post_iteration = async || -> Result<(), Box> { - let next_quote = cycle - .ask(FetchQuote) - .await - .map_err(|_| "fetch quote should always succeed")?; + // We store the code to perform the next posting iteration as one atomic future which we wrap with a timeout. + // This means that, if we miss a posting window due to the timeout, we will not get multiple consecutive or late posts. + let next_post_iteration = async || -> Result<(), Box> { + let next_quote = cycle + .ask(FetchQuote) + .await + .map_err(|_| "fetch quote should always succeed")?; - // Note: By using `tell`, we don't know when each sink's code will have completed. - // If any sink uses, say, a file or stdout, that resource may well be contested between - // consecutive iterations of this loop. - sink.tell(PostQuote(next_quote)).await?; - println!(); + // Note: By using `tell`, we don't know when each sink's code will have completed. + // If any sink uses, say, a file or stdout, that resource may well be contested between + // consecutive iterations of this loop. + sink.tell(PostQuote(next_quote)).await?; + println!(); - Ok(()) - }; + Ok(()) + }; - if let Err(e) = timeout(POSTING_TIMEOUT, next_post_iteration()).await { - eprintln!( - "Could not submit post in time to all sinks. Timeout error: {}", - e - ); - } - } + if let Err(e) = timeout(POSTING_TIMEOUT, next_post_iteration()).await { + eprintln!( + "Could not submit post in time to all sinks. Timeout error: {}", + e + ); + } + } - Ok(()) - } + Ok(()) + } } diff --git a/src/main.rs b/src/main.rs index 44e3614..904384a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,4 @@ #[tokio::main] async fn main() -> Result<(), Box> { - audquotes::run::entrypoint().await + audquotes::run::entrypoint().await } diff --git a/src/sink.rs b/src/sink.rs index 92711c9..53f62c2 100644 --- a/src/sink.rs +++ b/src/sink.rs @@ -13,20 +13,20 @@ pub struct PostQuote(pub Quote); /// behavior which other modules will observe. #[derive(Debug, Clone)] pub enum PostFailure { - /// Indicates that a given quote could not be posted to a sink, - /// but that it *may* be retried. The `reinitialize` boolean signals - /// whether the sink should be reinitialized before further attempts. - Retry { reinitialize: bool }, - - /// Indicates that a given quote could not be posted to a sink, - /// as it is unsupported by it in some way (e.g. quote exceeds the sink's length limit). - Unsupported, - - /// Indicates that a given quote could not be posted to a sink - /// due to the occurrence of some unrecoverable error. - /// It is thus unlikely that the sink will work in the future, even if - /// reinitialized. - Unrecoverable, + /// Indicates that a given quote could not be posted to a sink, + /// but that it *may* be retried. The `reinitialize` boolean signals + /// whether the sink should be reinitialized before further attempts. + Retry { reinitialize: bool }, + + /// Indicates that a given quote could not be posted to a sink, + /// as it is unsupported by it in some way (e.g. quote exceeds the sink's length limit). + Unsupported, + + /// Indicates that a given quote could not be posted to a sink + /// due to the occurrence of some unrecoverable error. + /// It is thus unlikely that the sink will work in the future, even if + /// reinitialized. + Unrecoverable, } pub type PostResult = Result<(), PostFailure>; @@ -41,78 +41,78 @@ pub trait QuoteSink: Actor + Message {} pub struct StdoutSink; impl Message for StdoutSink { - type Reply = PostResult; - - async fn handle( - &mut self, - PostQuote(quote): PostQuote, - _ctx: &mut Context, - ) -> Self::Reply { - println!("{}", quote.get()); - Ok(()) - } + type Reply = PostResult; + + async fn handle( + &mut self, + PostQuote(quote): PostQuote, + _ctx: &mut Context, + ) -> Self::Reply { + println!("{}", quote.get()); + Ok(()) + } } /// A [QuoteSink] which will post the contents of each quote to Bluesky. #[derive(Actor)] pub struct BskySink { - bsky_agent: BskyAgent, - bsky_session: Object, + bsky_agent: BskyAgent, + bsky_session: Object, } impl BskySink { - pub async fn new_session(username: String, password: String) -> Result { - let agent = BskyAgent::builder().build().await.map_err(|_| ())?; - let session = agent.login(username, password).await.map_err(|_| ())?; - - Ok(Self { - bsky_agent: agent, - bsky_session: session, - }) - } - - async fn submit_post(&mut self, quote: Quote) -> Result<(), ()> { - let post = bsky_sdk::api::app::bsky::feed::post::RecordData { - text: quote.into(), - created_at: bsky_sdk::api::types::string::Datetime::now(), - embed: None, - entities: None, - facets: None, - labels: None, - langs: None, - reply: None, - tags: None, - }; - - if let Err(e) = self - .bsky_agent - .resume_session(self.bsky_session.clone()) - .await - { - eprintln!("Failed to resume sessions due to following error: {e}"); - return Err(()); - } - - match self.bsky_agent.create_record(post.clone()).await { - Ok(_) => Ok(()), - Err(_) => Err(()), - } - } + pub async fn new_session(username: String, password: String) -> Result { + let agent = BskyAgent::builder().build().await.map_err(|_| ())?; + let session = agent.login(username, password).await.map_err(|_| ())?; + + Ok(Self { + bsky_agent: agent, + bsky_session: session, + }) + } + + async fn submit_post(&mut self, quote: Quote) -> Result<(), ()> { + let post = bsky_sdk::api::app::bsky::feed::post::RecordData { + text: quote.into(), + created_at: bsky_sdk::api::types::string::Datetime::now(), + embed: None, + entities: None, + facets: None, + labels: None, + langs: None, + reply: None, + tags: None, + }; + + if let Err(e) = self + .bsky_agent + .resume_session(self.bsky_session.clone()) + .await + { + eprintln!("Failed to resume sessions due to following error: {e}"); + return Err(()); + } + + match self.bsky_agent.create_record(post.clone()).await { + Ok(_) => Ok(()), + Err(_) => Err(()), + } + } } impl Message for BskySink { - type Reply = PostResult; - - async fn handle( - &mut self, - PostQuote(quote): PostQuote, - _ctx: &mut Context, - ) -> Self::Reply { - match self.submit_post(quote).await { - Ok(_) => Ok(()), - Err(_) => Err(PostFailure::Unrecoverable), - } - } + type Reply = PostResult; + + async fn handle( + &mut self, + PostQuote(quote): PostQuote, + _ctx: &mut Context, + ) -> Self::Reply { + match self.submit_post(quote).await { + Ok(_) => Ok(()), + Err(_) => Err(PostFailure::Unrecoverable), + } + } } /// Supervises all [QuoteSink] actors within the program, forwarding @@ -121,72 +121,72 @@ impl Message for BskySink { /// encountering recoverable errors. #[derive(Actor)] pub struct SinkManager { - // Uh oh. As the [Actor] trait is *not* dyn-compatible, - // and I do not own its definition, I'm fairly certain that I cannot - // do asynchronous dynamic dispatch for it here. - // I've decided I'll limit this to one sink per implementation right now. - stdout_sink: Option>, - bsky_sink: Option>, - // ... + // Uh oh. As the [Actor] trait is *not* dyn-compatible, + // and I do not own its definition, I'm fairly certain that I cannot + // do asynchronous dynamic dispatch for it here. + // I've decided I'll limit this to one sink per implementation right now. + stdout_sink: Option>, + bsky_sink: Option>, + // ... } impl SinkManager { - pub fn new( - stdout_sink: Option>, - bsky_sink: Option>, - ) -> Self { - Self { - stdout_sink, - bsky_sink, - } - } + pub fn new( + stdout_sink: Option>, + bsky_sink: Option>, + ) -> Self { + Self { + stdout_sink, + bsky_sink, + } + } } pub type SinkReplies = Vec>; impl Message for SinkManager { - type Reply = SinkReplies; - - async fn handle( - &mut self, - msg: PostQuote, - _ctx: &mut Context, - ) -> Self::Reply { - use futures::future::join_all; - - let stdout_result = self - .stdout_sink - .as_ref() - .map(|s| s.ask(msg.clone()).into_future()); - - let bsky_result = self - .bsky_sink - .as_ref() - .map(|s| s.ask(msg.clone()).into_future()); - - let futures = [stdout_result, bsky_result].into_iter().flatten(); - let results = join_all(futures).await; - - results.iter().map(|r| r.clone().or(Err(()))).collect() - } + type Reply = SinkReplies; + + async fn handle( + &mut self, + msg: PostQuote, + _ctx: &mut Context, + ) -> Self::Reply { + use futures::future::join_all; + + let stdout_result = self + .stdout_sink + .as_ref() + .map(|s| s.ask(msg.clone()).into_future()); + + let bsky_result = self + .bsky_sink + .as_ref() + .map(|s| s.ask(msg.clone()).into_future()); + + let futures = [stdout_result, bsky_result].into_iter().flatten(); + let results = join_all(futures).await; + + results.iter().map(|r| r.clone().or(Err(()))).collect() + } } mod test { - #[tokio::test] - async fn stdout_sink() { - use super::*; - - let stdout = StdoutSink::spawn(StdoutSink); - let manager = SinkManager::spawn(SinkManager::new(Some(stdout), None)); - - let messages = ["First test!", "Second test.", "Third..."]; - for msg in messages { - manager.tell(PostQuote(msg.into())).await.unwrap(); - tokio::time::sleep(std::time::Duration::from_secs(5)).await; - } - - // Hopefully we don't crash...! - // TODO: Sink that actually stores every quote it "posts"? - // Could help in verifying everything was sent correctly. - } + #[tokio::test] + async fn stdout_sink() { + use super::*; + + let stdout = StdoutSink::spawn(StdoutSink); + let manager = SinkManager::spawn(SinkManager::new(Some(stdout), None)); + + let messages = ["First test!", "Second test.", "Third..."]; + for msg in messages { + manager.tell(PostQuote(msg.into())).await.unwrap(); + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + } + + // Hopefully we don't crash...! + // TODO: Sink that actually stores every quote it "posts"? + // Could help in verifying everything was sent correctly. + } } diff --git a/src/storage.rs b/src/storage.rs index bf1872d..24be6aa 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -1,279 +1,279 @@ use kameo::prelude::*; use crate::storage::{ - queue::{DequeueQuote, EnqueueQuotes}, - source::SourceQuotes, + queue::{DequeueQuote, EnqueueQuotes}, + source::SourceQuotes, }; use crate::data::Quote; mod rng { - use rand::SeedableRng; - - pub struct PrngState { - rng: rand::rngs::SmallRng, - } - - impl PrngState { - pub fn from_thread_rng() -> Self { - Self { - rng: rand::rngs::SmallRng::from_rng(&mut rand::rng()), - } - } - - pub fn from_seed(seed: u64) -> Self { - Self { - rng: rand::rngs::SmallRng::seed_from_u64(seed), - } - } - - pub fn shuffle_slice(&mut self, slice: &mut [T]) { - use rand::seq::SliceRandom; - slice.shuffle(&mut self.rng); - } - } - - mod test { - #[tokio::test] - async fn shuffle_slice() { - use super::*; - - let mut data = vec![1, 2, 3, 4]; - let mut rng = PrngState::from_thread_rng(); - - rng.shuffle_slice(&mut data); - println!("{:?}", data); - } - } + use rand::SeedableRng; + + pub struct PrngState { + rng: rand::rngs::SmallRng, + } + + impl PrngState { + pub fn from_thread_rng() -> Self { + Self { + rng: rand::rngs::SmallRng::from_rng(&mut rand::rng()), + } + } + + pub fn from_seed(seed: u64) -> Self { + Self { + rng: rand::rngs::SmallRng::seed_from_u64(seed), + } + } + + pub fn shuffle_slice(&mut self, slice: &mut [T]) { + use rand::seq::SliceRandom; + slice.shuffle(&mut self.rng); + } + } + + mod test { + #[tokio::test] + async fn shuffle_slice() { + use super::*; + + let mut data = vec![1, 2, 3, 4]; + let mut rng = PrngState::from_thread_rng(); + + rng.shuffle_slice(&mut data); + println!("{:?}", data); + } + } } pub mod source { - use super::*; - - // TODO: Should the quote source filters be - // generic over the exact manager implementation being used? - /// Message to request that a SourceManager source its quotes once again. - pub struct SourceQuotes; - pub type SourceReply = Result, ()>; - - /// Subtrait of Actor which specifically - /// denotes actors that can handle all relevant source messages. - pub trait SourceManager: Actor + Message {} - - impl SourceManager for T where T: Message {} - - /// Implementation of [`SourceManager`] which sources quotes from a Vec - /// that it holds in memory, without accessing external services. - /// Its main purpose is to be used for testing. - #[derive(Actor)] - pub struct MemorySourceManager { - quotes: Vec, - } - - impl MemorySourceManager { - pub fn new(quotes: impl IntoIterator>) -> Self { - Self { - quotes: quotes.into_iter().map(Into::into).collect(), - } - } - } - - impl Message for MemorySourceManager { - type Reply = SourceReply; - - async fn handle( - &mut self, - _msg: SourceQuotes, - _ctx: &mut Context, - ) -> Self::Reply { - // We just clone the quotes we've been holding onto since startup - Ok(self.quotes.clone()) - } - } - - /// Uses a [QuoteFilter] to source quotes from the local filesystem - /// at the beginning of each cycle. - #[derive(Actor)] - pub struct FsFilterSourceManager { - filter: QuoteFilter, - } - - impl FsFilterSourceManager { - pub fn new(filter: QuoteFilter) -> Self { - Self { filter } - } - } - - impl Default for FsFilterSourceManager { - fn default() -> Self { - Self { - filter: QuoteFilter { - content: r".*".to_string(), - // TODO: Maybe make this a compile-time constant for debugging - path: "test/**/*.txt".to_string(), - _dates: vec![], - }, - } - } - } - - impl Message for FsFilterSourceManager { - type Reply = SourceReply; - - async fn handle( - &mut self, - _msg: SourceQuotes, - _ctx: &mut Context, - ) -> Self::Reply { - self.filter.read_files() - } - } - - #[derive(Clone, Debug)] - pub struct QuoteFilter { - path: String, - content: String, - _dates: Vec, - } - - impl QuoteFilter { - // TODO: actually leverage async I/O - fn read_files(&self) -> Result, ()> { - use glob::glob; - use grep::{regex, searcher::sinks}; - - let matcher = regex::RegexMatcher::new(&self.content).map_err(|_| ())?; - let mut searcher = grep::searcher::Searcher::new(); - let mut results = Vec::new(); - - for file in glob(&self.path).map_err(|_| ())? { - let file = match file { - Ok(file) => file, - Err(_) => continue, - }; - - let mut matched = false; - let sink = sinks::Lossy(|_lnum, _line| { - matched = true; - Ok(false) - }); - - let search_result = searcher.search_path(&matcher, &file, sink); - if !matched || search_result.is_err() { - continue; - } - - let contents = std::fs::read_to_string(file).map_err(|_| ())?; - results.push(contents.trim().into()); - } - - Ok(results) - } - } + use super::*; + + // TODO: Should the quote source filters be + // generic over the exact manager implementation being used? + /// Message to request that a SourceManager source its quotes once again. + pub struct SourceQuotes; + pub type SourceReply = Result, ()>; + + /// Subtrait of Actor which specifically + /// denotes actors that can handle all relevant source messages. + pub trait SourceManager: Actor + Message {} + + impl SourceManager for T where T: Message {} + + /// Implementation of [`SourceManager`] which sources quotes from a Vec + /// that it holds in memory, without accessing external services. + /// Its main purpose is to be used for testing. + #[derive(Actor)] + pub struct MemorySourceManager { + quotes: Vec, + } + + impl MemorySourceManager { + pub fn new(quotes: impl IntoIterator>) -> Self { + Self { + quotes: quotes.into_iter().map(Into::into).collect(), + } + } + } + + impl Message for MemorySourceManager { + type Reply = SourceReply; + + async fn handle( + &mut self, + _msg: SourceQuotes, + _ctx: &mut Context, + ) -> Self::Reply { + // We just clone the quotes we've been holding onto since startup + Ok(self.quotes.clone()) + } + } + + /// Uses a [QuoteFilter] to source quotes from the local filesystem + /// at the beginning of each cycle. + #[derive(Actor)] + pub struct FsFilterSourceManager { + filter: QuoteFilter, + } + + impl FsFilterSourceManager { + pub fn new(filter: QuoteFilter) -> Self { + Self { filter } + } + } + + impl Default for FsFilterSourceManager { + fn default() -> Self { + Self { + filter: QuoteFilter { + content: r".*".to_string(), + // TODO: Maybe make this a compile-time constant for debugging + path: "test/**/*.txt".to_string(), + _dates: vec![], + }, + } + } + } + + impl Message for FsFilterSourceManager { + type Reply = SourceReply; + + async fn handle( + &mut self, + _msg: SourceQuotes, + _ctx: &mut Context, + ) -> Self::Reply { + self.filter.read_files() + } + } + + #[derive(Clone, Debug)] + pub struct QuoteFilter { + path: String, + content: String, + _dates: Vec, + } + + impl QuoteFilter { + // TODO: actually leverage async I/O + fn read_files(&self) -> Result, ()> { + use glob::glob; + use grep::{regex, searcher::sinks}; + + let matcher = regex::RegexMatcher::new(&self.content).map_err(|_| ())?; + let mut searcher = grep::searcher::Searcher::new(); + let mut results = Vec::new(); + + for file in glob(&self.path).map_err(|_| ())? { + let file = match file { + Ok(file) => file, + Err(_) => continue, + }; + + let mut matched = false; + let sink = sinks::Lossy(|_lnum, _line| { + matched = true; + Ok(false) + }); + + let search_result = searcher.search_path(&matcher, &file, sink); + if !matched || search_result.is_err() { + continue; + } + + let contents = std::fs::read_to_string(file).map_err(|_| ())?; + results.push(contents.trim().into()); + } + + Ok(results) + } + } } pub mod queue { - use std::collections::VecDeque; - - use super::*; - - // Messages to interact with the quote queue - pub struct DequeueQuote; - pub type DequeueReply = Result, ()>; - - pub struct EnqueueQuotes(pub Vec); - pub type EnqueueReply = Result<(), ()>; - - /// Subtrait of Actor which specifically - /// denotes actors that can handle all relevant queue messages. - pub trait QueueManager: - Actor - + Message - + Message - { - } - - impl QueueManager for T where - T: Message - + Message - { - } - - /// A basic implementation of an in-memory queue of quotes. - /// Its contents are *not* persisted across application restarts, so it - /// is only suited for testing purposes. - #[derive(Actor, Default)] - pub struct MemoryQueueStorage { - quotes: VecDeque, - } - - impl MemoryQueueStorage { - pub fn new() -> Self { - Self { - quotes: VecDeque::new(), - } - } - } - - impl Message for MemoryQueueStorage { - /// We only need to signal success or failure in this instance, - /// with no added metadata in either case. - type Reply = EnqueueReply; - - async fn handle( - &mut self, - msg: EnqueueQuotes, - _ctx: &mut Context, - ) -> Self::Reply { - for q in msg.0 { - self.quotes.push_back(q); - } - - Ok(()) - } - } - - impl Message for MemoryQueueStorage { - type Reply = DequeueReply; - - async fn handle( - &mut self, - _msg: DequeueQuote, - _ctx: &mut Context, - ) -> Self::Reply { - // Note: this can never fail, since the quotes are stored in memory - Ok(self.quotes.pop_front()) - } - } + use std::collections::VecDeque; + + use super::*; + + // Messages to interact with the quote queue + pub struct DequeueQuote; + pub type DequeueReply = Result, ()>; + + pub struct EnqueueQuotes(pub Vec); + pub type EnqueueReply = Result<(), ()>; + + /// Subtrait of Actor which specifically + /// denotes actors that can handle all relevant queue messages. + pub trait QueueManager: + Actor + + Message + + Message + { + } + + impl QueueManager for T where + T: Message + + Message + { + } + + /// A basic implementation of an in-memory queue of quotes. + /// Its contents are *not* persisted across application restarts, so it + /// is only suited for testing purposes. + #[derive(Actor, Default)] + pub struct MemoryQueueStorage { + quotes: VecDeque, + } + + impl MemoryQueueStorage { + pub fn new() -> Self { + Self { + quotes: VecDeque::new(), + } + } + } + + impl Message for MemoryQueueStorage { + /// We only need to signal success or failure in this instance, + /// with no added metadata in either case. + type Reply = EnqueueReply; + + async fn handle( + &mut self, + msg: EnqueueQuotes, + _ctx: &mut Context, + ) -> Self::Reply { + for q in msg.0 { + self.quotes.push_back(q); + } + + Ok(()) + } + } + + impl Message for MemoryQueueStorage { + type Reply = DequeueReply; + + async fn handle( + &mut self, + _msg: DequeueQuote, + _ctx: &mut Context, + ) -> Self::Reply { + // Note: this can never fail, since the quotes are stored in memory + Ok(self.quotes.pop_front()) + } + } } #[derive(Actor)] pub struct QuoteCycle { - rng: rng::PrngState, - source_manager: ActorRef, - queue_manager: ActorRef, + rng: rng::PrngState, + source_manager: ActorRef, + queue_manager: ActorRef, } impl QuoteCycle { - pub fn new( - rng: rng::PrngState, - source_manager: ActorRef, - queue_manager: ActorRef, - ) -> Self { - Self { - rng, - source_manager, - queue_manager, - } - } - - pub fn with_thread_rng(source_manager: ActorRef, queue_manager: ActorRef) -> Self { - Self { - rng: rng::PrngState::from_thread_rng(), - source_manager, - queue_manager, - } - } + pub fn new( + rng: rng::PrngState, + source_manager: ActorRef, + queue_manager: ActorRef, + ) -> Self { + Self { + rng, + source_manager, + queue_manager, + } + } + + pub fn with_thread_rng(source_manager: ActorRef, queue_manager: ActorRef) -> Self { + Self { + rng: rng::PrngState::from_thread_rng(), + source_manager, + queue_manager, + } + } } /// A message to [QuoteCycle] to fetch one more quote from its storage. @@ -281,144 +281,144 @@ pub struct FetchQuote; impl Message for QuoteCycle where - S: source::SourceManager, - Q: queue::QueueManager, + S: source::SourceManager, + Q: queue::QueueManager, { - type Reply = Result; - - async fn handle( - &mut self, - _msg: FetchQuote, - _ctx: &mut Context, - ) -> Self::Reply { - // 1. We query our queue storage for the next quote - if let Some(next_quote) = self.queue_manager.ask(DequeueQuote).await.map_err(|_| ())? { - // if there is a quote, we simply return it and move on - return Ok(next_quote); - } - - // 2. Otherwise, we must repopulate the queue through our source - let mut refreshed_quotes = self - .source_manager - .ask(SourceQuotes) - .await - .map_err(|_| ())?; - - // 3. We shuffle the newly-sourced quotes - self.rng.shuffle_slice(&mut refreshed_quotes); - let refreshed_quotes = refreshed_quotes; // No longer mutable - - // TODO: Perhaps we should assert that the new quotes are non-empty? - // 4. We enqueue the newly-sourced quotes... - self.queue_manager - .ask(EnqueueQuotes(refreshed_quotes)) - .await - .map_err(|_| ())?; - - // 5. and, finally, we return the first among them. - match self.queue_manager.ask(DequeueQuote).await { - Ok(Some(q)) => Ok(q), - Ok(None) => panic!("Newly-enqueued quotes should never be empty"), - Err(_) => Err(()), - } - } + type Reply = Result; + + async fn handle( + &mut self, + _msg: FetchQuote, + _ctx: &mut Context, + ) -> Self::Reply { + // 1. We query our queue storage for the next quote + if let Some(next_quote) = self.queue_manager.ask(DequeueQuote).await.map_err(|_| ())? { + // if there is a quote, we simply return it and move on + return Ok(next_quote); + } + + // 2. Otherwise, we must repopulate the queue through our source + let mut refreshed_quotes = self + .source_manager + .ask(SourceQuotes) + .await + .map_err(|_| ())?; + + // 3. We shuffle the newly-sourced quotes + self.rng.shuffle_slice(&mut refreshed_quotes); + let refreshed_quotes = refreshed_quotes; // No longer mutable + + // TODO: Perhaps we should assert that the new quotes are non-empty? + // 4. We enqueue the newly-sourced quotes... + self.queue_manager + .ask(EnqueueQuotes(refreshed_quotes)) + .await + .map_err(|_| ())?; + + // 5. and, finally, we return the first among them. + match self.queue_manager.ask(DequeueQuote).await { + Ok(Some(q)) => Ok(q), + Ok(None) => panic!("Newly-enqueued quotes should never be empty"), + Err(_) => Err(()), + } + } } mod test { - #[tokio::test] - async fn memory_queue() { - use super::Quote; - use super::queue::*; - use kameo::prelude::*; - - let queue_manager = MemoryQueueStorage::spawn(MemoryQueueStorage::new()); - - let sample_quotes = ["Test no.1", "Test no.2", "Test no.3"]; - - queue_manager - .ask(EnqueueQuotes( - sample_quotes.iter().cloned().map(Quote::from).collect(), - )) - .await - .expect("In-memory quote queue storage should be valid for insertion"); - - for text in sample_quotes.iter() { - assert_eq!( - *text, - queue_manager - .ask(DequeueQuote) - .await - .expect("In-memory queue storage should never panic on dequeue") - .expect("In-memory queue storage should never be initialized as empty") - .get() - ); - } - } - - #[tokio::test] - async fn memory_source() { - use super::source::*; - use kameo::prelude::*; - - let sample_quotes = ["Minie", "Miney", "Moe", "and", "some", "more"]; - - let source_manager = MemorySourceManager::spawn(MemorySourceManager::new(sample_quotes)); - - let quotes = source_manager - .ask(SourceQuotes) - .await - .expect("In-memory quote queue storage should be valid for insertion"); - - assert_eq!( - sample_quotes.as_slice(), - quotes - .into_iter() - // Since [Quote] doesn't implement any Equality trait, - // we map the strings into quotes instead - .map(String::from) - .collect::>() - .as_slice(), - ); - } - - #[tokio::test] - async fn memory_cycle() { - use std::{collections::HashMap, ops::AddAssign}; - - use super::FetchQuote; - use super::QuoteCycle; - use super::queue::*; - use super::source::*; - use kameo::prelude::*; - - let sample_quotes = ["Minie", "Miney", "Moe"]; - let cycle = { - let source = MemorySourceManager::spawn(MemorySourceManager::new(sample_quotes)); - let queue = MemoryQueueStorage::spawn(MemoryQueueStorage::new()); - - QuoteCycle::spawn(QuoteCycle::with_thread_rng(source, queue)) - }; - - // We loop over `sample_quotes` twice to simulate the queue being exhausted fully, then re-sourced - // Since the `cycle` manager will shuffle the quote sequence, we will verify that each - // quote appears *exactly* `LOOPS` times throughout these iterations. - const LOOPS: usize = 3; - let mut quote_counts = HashMap::new(); - for _ in 0..(sample_quotes.len() * LOOPS) { - let next_quote = cycle.ask(FetchQuote).await.unwrap(); - quote_counts - .entry(next_quote.get().to_owned()) - .or_insert(0) - .add_assign(1); - } - - let quote_counts = quote_counts; // no longer mut - assert!( - // Note: technically speaking, different quotes could contain equivalent strings, - // which would make this test fail; a "more proper" invariant check would ensure - // verify that all counts are a multiple of the amount of times `sample_quotes` was chained, - // and that the sum of all counts equals the total number of times a `FetchQuote` message was sent. - quote_counts.into_values().into_iter().all(|c| c == LOOPS) - ); - } + #[tokio::test] + async fn memory_queue() { + use super::Quote; + use super::queue::*; + use kameo::prelude::*; + + let queue_manager = MemoryQueueStorage::spawn(MemoryQueueStorage::new()); + + let sample_quotes = ["Test no.1", "Test no.2", "Test no.3"]; + + queue_manager + .ask(EnqueueQuotes( + sample_quotes.iter().cloned().map(Quote::from).collect(), + )) + .await + .expect("In-memory quote queue storage should be valid for insertion"); + + for text in sample_quotes.iter() { + assert_eq!( + *text, + queue_manager + .ask(DequeueQuote) + .await + .expect("In-memory queue storage should never panic on dequeue") + .expect("In-memory queue storage should never be initialized as empty") + .get() + ); + } + } + + #[tokio::test] + async fn memory_source() { + use super::source::*; + use kameo::prelude::*; + + let sample_quotes = ["Minie", "Miney", "Moe", "and", "some", "more"]; + + let source_manager = MemorySourceManager::spawn(MemorySourceManager::new(sample_quotes)); + + let quotes = source_manager + .ask(SourceQuotes) + .await + .expect("In-memory quote queue storage should be valid for insertion"); + + assert_eq!( + sample_quotes.as_slice(), + quotes + .into_iter() + // Since [Quote] doesn't implement any Equality trait, + // we map the strings into quotes instead + .map(String::from) + .collect::>() + .as_slice(), + ); + } + + #[tokio::test] + async fn memory_cycle() { + use std::{collections::HashMap, ops::AddAssign}; + + use super::FetchQuote; + use super::QuoteCycle; + use super::queue::*; + use super::source::*; + use kameo::prelude::*; + + let sample_quotes = ["Minie", "Miney", "Moe"]; + let cycle = { + let source = MemorySourceManager::spawn(MemorySourceManager::new(sample_quotes)); + let queue = MemoryQueueStorage::spawn(MemoryQueueStorage::new()); + + QuoteCycle::spawn(QuoteCycle::with_thread_rng(source, queue)) + }; + + // We loop over `sample_quotes` twice to simulate the queue being exhausted fully, then re-sourced + // Since the `cycle` manager will shuffle the quote sequence, we will verify that each + // quote appears *exactly* `LOOPS` times throughout these iterations. + const LOOPS: usize = 3; + let mut quote_counts = HashMap::new(); + for _ in 0..(sample_quotes.len() * LOOPS) { + let next_quote = cycle.ask(FetchQuote).await.unwrap(); + quote_counts + .entry(next_quote.get().to_owned()) + .or_insert(0) + .add_assign(1); + } + + let quote_counts = quote_counts; // no longer mut + assert!( + // Note: technically speaking, different quotes could contain equivalent strings, + // which would make this test fail; a "more proper" invariant check would ensure + // verify that all counts are a multiple of the amount of times `sample_quotes` was chained, + // and that the sum of all counts equals the total number of times a `FetchQuote` message was sent. + quote_counts.into_values().into_iter().all(|c| c == LOOPS) + ); + } } -- 2.51.2