notify/examples/watch.rs

use notify::{Listener, Notify};
use std::{
    sync::{
        atomic::{AtomicBool, AtomicUsize, Ordering},
        Arc, Mutex,
    },
    time::Duration,
};

struct State<T> {
    item: Mutex<T>,
    current_id: AtomicUsize,
    tx_dropped: AtomicBool,
}

pub struct Sender<T> {
    state: Arc<State<T>>,
    notify: Notify,
}

pub struct Receiver<T> {
    state: Arc<State<T>>,
    listener: Listener,
    last_seen: usize,
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    jive::block_on(async move {
        let (tx, rx) = watch("meowdy".into());

        let mut handles = Vec::new();
        handles.push(jive::spawn(async move {
            let mut interval = jive::time::interval(Duration::from_secs(2));

            for i in 0..5 {
                interval.tick().await;
                let _ = tx.send(format!("hewwo {}", i));
            }
        }));

        for i in 0..5 {
            let mut rx = rx.clone();
            handles.push(jive::spawn(async move {
                while let Some(item) = rx.recv().await {
                    println!("{} Got: {}", i, item);
                }
            }));
        }

        drop(rx);

        for handle in handles {
            let _ = handle.await;
        }
    });

    let (tx, rx) = watch("meowdy".into());

    let mut handles = Vec::new();
    handles.push(std::thread::spawn(move || {
        for i in 0..5 {
            std::thread::sleep(Duration::from_secs(2));
            let _ = tx.send(format!("meowdy {}", i));
        }
    }));

    for i in 0..5 {
        let mut rx = rx.clone();
        handles.push(std::thread::spawn(move || {
            while let Some(item) = rx.recv_blocking() {
                println!("{} Got: {}", i, item);
            }
        }));
    }

    drop(rx);

    for handle in handles {
        let _ = handle.join();
    }

    Ok(())
}

pub fn watch<T>(item: T) -> (Sender<T>, Receiver<T>) {
    let state = Arc::new(State {
        item: Mutex::new(item),
        current_id: AtomicUsize::new(0),
        tx_dropped: AtomicBool::new(false),
    });
    let notify = Notify::new();
    let listener = notify.listener();

    (
        Sender {
            state: Arc::clone(&state),
            notify,
        },
        Receiver {
            state,
            listener,
            last_seen: 0,
        },
    )
}

impl<T> Sender<T> {
    pub fn send(&self, item: T) -> Result<(), T> {
        if self.is_closed() {
            return Err(item);
        }

        self.notify.listener().consume_all_notifications();
        *self.state.item.lock().unwrap() = item;
        self.state.current_id.fetch_add(1, Ordering::AcqRel);
        self.notify.notify(usize::MAX);

        Ok(())
    }

    pub fn is_closed(&self) -> bool {
        Arc::strong_count(&self.state) == 1
    }
}

impl<T> Receiver<T> {
    pub fn try_recv(&mut self) -> Option<T>
    where
        T: Clone,
    {
        let current_id = self.state.current_id.load(Ordering::Acquire);
        if current_id == self.last_seen {
            return None;
        }

        self.last_seen = current_id;

        Some(self.state.item.lock().unwrap().clone())
    }

    pub async fn recv(&mut self) -> Option<T>
    where
        T: Clone,
    {
        loop {
            if let Some(item) = self.try_recv() {
                return Some(item);
            }

            if self.is_closed() {
                return None;
            }

            self.listener.listen().await;
        }
    }

    pub fn recv_blocking(&mut self) -> Option<T>
    where
        T: Clone,
    {
        loop {
            if let Some(item) = self.try_recv() {
                return Some(item);
            }

            if self.is_closed() {
                return None;
            }

            self.listener.listen_blocking();
        }
    }

    pub fn is_closed(&self) -> bool {
        self.state.tx_dropped.load(Ordering::Acquire)
    }
}

impl<T> Clone for Receiver<T> {
    fn clone(&self) -> Self {
        Self {
            state: Arc::clone(&self.state),
            listener: self.listener.clone(),
            last_seen: self.last_seen,
        }
    }
}

impl<T> Drop for Sender<T> {
    fn drop(&mut self) {
        self.state.tx_dropped.store(true, Ordering::Release);
    }
}