Quelle  statecallback.rs   Sprache: Rust

/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

use std::sync::{Arc, Condvar, Mutex};

pub struct StateCallback<T> {
    callback: Arc<Mutex<Option<Box<dyn FnOnce(T) + Send>>>>,
    observer: Arc<Mutex<Option<Box<dyn FnOnce() + Send>>>>,
    condition: Arc<(Mutex<bool>, Condvar)>,
}

impl<T> StateCallback<T> {
    // This is used for the Condvar, which requires this kind of construction
    #[allow(clippy::mutex_atomic)]
    pub fn new(cb: Box<dyn FnOnce(T) + Send>) -> Self {
        Self {
            callback: Arc::new(Mutex::new(Some(cb))),
            observer: Arc::new(Mutex::new(None)),
            condition: Arc::new((Mutex::new(true), Condvar::new())),
        }
    }

    pub fn add_uncloneable_observer(&mut self, obs: Box<dyn FnOnce() + Send>) {
        let mut opt = self.observer.lock().unwrap();
        if opt.is_some() {
            error!("Replacing an already-set observer.")
        }
        opt.replace(obs);
    }

    pub fn call(&self, rv: T) {
        if let Some(cb) = self.callback.lock().unwrap().take() {
            cb(rv);

            if let Some(obs) = self.observer.lock().unwrap().take() {
                obs();
            }
        }

        let (lock, cvar) = &*self.condition;
        let mut pending = lock.lock().unwrap();
        *pending = false;
        cvar.notify_all();
    }

    pub fn wait(&self) {
        let (lock, cvar) = &*self.condition;
        let _useless_guard = cvar
            .wait_while(lock.lock().unwrap(), |pending| *pending)
            .unwrap();
    }
}

impl<T> Clone for StateCallback<T> {
    fn clone(&self) -> Self {
        Self {
            callback: self.callback.clone(),
            observer: Arc::new(Mutex::new(None)),
            condition: self.condition.clone(),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::StateCallback;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Arc, Barrier};
    use std::thread;

    #[test]
    fn test_statecallback_is_single_use() {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = counter.clone();
        let sc = StateCallback::new(Box::new(move |_| {
            counter_clone.fetch_add(1, Ordering::SeqCst);
        }));

        assert_eq!(counter.load(Ordering::SeqCst), 0);
        for _ in 0..10 {
            sc.call(());
            assert_eq!(counter.load(Ordering::SeqCst), 1);
        }

        for _ in 0..10 {
            sc.clone().call(());
            assert_eq!(counter.load(Ordering::SeqCst), 1);
        }
    }

    #[test]
    fn test_statecallback_observer_is_single_use() {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = counter.clone();
        let mut sc = StateCallback::<()>::new(Box::new(move |_| {}));

        sc.add_uncloneable_observer(Box::new(move || {
            counter_clone.fetch_add(1, Ordering::SeqCst);
        }));

        assert_eq!(counter.load(Ordering::SeqCst), 0);
        for _ in 0..10 {
            sc.call(());
            assert_eq!(counter.load(Ordering::SeqCst), 1);
        }

        for _ in 0..10 {
            sc.clone().call(());
            assert_eq!(counter.load(Ordering::SeqCst), 1);
        }
    }

    #[test]
    fn test_statecallback_observer_only_runs_for_completing_callback() {
        let cb_counter = Arc::new(AtomicUsize::new(0));
        let cb_counter_clone = cb_counter.clone();
        let sc = StateCallback::new(Box::new(move |_| {
            cb_counter_clone.fetch_add(1, Ordering::SeqCst);
        }));

        let obs_counter = Arc::new(AtomicUsize::new(0));

        for _ in 0..10 {
            let obs_counter_clone = obs_counter.clone();
            let mut c = sc.clone();
            c.add_uncloneable_observer(Box::new(move || {
                obs_counter_clone.fetch_add(1, Ordering::SeqCst);
            }));

            c.call(());

            assert_eq!(cb_counter.load(Ordering::SeqCst), 1);
            assert_eq!(obs_counter.load(Ordering::SeqCst), 1);
        }
    }

    #[test]
    #[allow(clippy::redundant_clone)]
    fn test_statecallback_observer_unclonable() {
        let mut sc = StateCallback::<()>::new(Box::new(move |_| {}));
        sc.add_uncloneable_observer(Box::new(move || {}));

        assert!(sc.observer.lock().unwrap().is_some());
        // This is deliberate, to force an extra clone
        assert!(sc.clone().observer.lock().unwrap().is_none());
    }

    #[test]
    fn test_statecallback_wait() {
        let sc = StateCallback::<()>::new(Box::new(move |_| {}));
        let barrier = Arc::new(Barrier::new(2));

        {
            let c = sc.clone();
            let b = barrier.clone();
            thread::spawn(move || {
                b.wait();
                c.call(());
            });
        }

        barrier.wait();
        sc.wait();
    }
}