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) + ); + } }