You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
genmarkov/src/msg.rs

163 lines
3.5 KiB

//! Message passing things
use super::*;
use tokio::{
sync::{
watch,
},
};
use std::{
task::{Poll, Context},
pin::Pin,
fmt,
error,
};
use futures::{
future::Future,
};
#[derive(Debug)]
pub struct InitError;
#[derive(Debug)]
pub struct InitWaitError;
impl error::Error for InitError{}
impl fmt::Display for InitError
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result
{
write!(f, "failed to set init value")
}
}
impl error::Error for InitWaitError{}
impl fmt::Display for InitWaitError
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result
{
write!(f, "failed to receive init value")
}
}
/// Provides a method of waiting on and setting a single initialisation.
///
/// In general, it should only be set once, as multiple sets do nothing but hog `Arc`s.
/// Dropping the `Initialiser` after waiting or setting should generally be done immediately.
/// Choose the `into_wait()` and `set()` varients over the non-consuming ones.
#[derive(Clone, Debug)]
pub struct Initialiser
{
tx: Arc<watch::Sender<bool>>,
rx: watch::Receiver<bool>
}
impl Initialiser
{
/// Create a new, unset initialiser
pub fn new() -> Self
{
let (tx, rx) = watch::channel(false);
Self {
tx: Arc::new(tx),
rx,
}
}
/// Create a pre-set initialiser. Calls to `wait()` will immediately resolve.
pub fn new_set() -> Self
{
let (tx, rx) = watch::channel(true);
Self {
tx: Arc::new(tx),
rx,
}
}
/// Consume into a future that completes when init is set.
pub fn into_wait(self) -> impl Future<Output=Result<(), InitWaitError>> + 'static
{
let mut rx = self.rx;
async move {
if !*rx.borrow() {
while !rx.recv().await.ok_or_else(|| InitWaitError)? {
//tokio::task::yield_now().await;
}
Ok(())
} else {
Ok(())
}
}
}
/// Clone into a future that completes when init is set.
///
/// This method does not clone any `Arc`s and is prefered to `self.clone().into_wait()`.
/// Use this when the `Initialiser` you want to wait on is behind a shared reference.
pub fn clone_into_wait(&self) -> impl Future<Output=Result<(), InitWaitError>> + 'static
{
let mut rx = self.rx.clone();
async move {
if !*rx.borrow() {
while !rx.recv().await.ok_or_else(|| InitWaitError)? {
//tokio::task::yield_now().await;
}
Ok(())
} else {
Ok(())
}
}
}
/// Completes when init is set
pub async fn wait(&mut self) -> Result<(), InitWaitError>
{
if !*self.rx.borrow() {
while !self.rx.recv().await.ok_or_else(|| InitWaitError)? {
//tokio::task::yield_now().await;
}
Ok(())
} else {
Ok(())
}
}
/// Is init set?
pub fn is_set(&self) -> bool
{
*self.rx.borrow()
}
/// Consume and set init if it's not already set
pub fn set(self) -> Result<(), InitError>
{
if !*self.rx.borrow() {
self.tx.broadcast(true).map_err(|_| InitError)
} else {
Ok(())
}
}
/// Set init without consuming.
///
/// # Note
/// It is prefered to use `set()`, as this method may make `Arc`s hang around longer than they need to.
/// Calling this multiple times is useless.
pub fn set_in_place(&self) -> Result<(), InitError>
{
if !*self.rx.borrow() {
self.tx.broadcast(true).map_err(|_| InitError)
} else {
Ok(())
}
}
}
impl Future for Initialiser
{
type Output = Result<(), InitWaitError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let uhh = self.wait();
tokio::pin!(uhh);
uhh.poll(cx)
}
}