retry.rs

#[macro_export]
macro_rules! retry {
    ( @create_multiple_runner $timeout:expr, $started_at:expr, $( $name:ident: $type:ty => $block:expr ),+ ) => {
        {
            let resources = Resources {
                $( 
                    $name: retry!(@create_runner $timeout, $started_at, $block)
                ),+
            };

            Ok((
                $(
                    match resources.$name.join()  {
                        Ok(Ok(val)) => val,
                        Ok(Err(_)) | Err(_) => return Err(())
                    }
                ),+
            ))
        }
    };

    ( @create_runner $timeout:expr, $started_at:expr, $block:expr ) => {
        {
            let timeout = $timeout.clone();
            let started_at = $started_at.clone();
            let closure = Arc::new($block);

            thread::spawn(move || {
                loop {
                    let t = timeout.clone();
                    let s = started_at.clone();

                    let f = closure.clone();
                    match thread::spawn(move || { f() }).join() {
                        Ok(Ok(val))  => return Ok(val),
                        Ok(Err(_)) | Err(_) => {
                            if s.elapsed() >= *t {
                                return Err(());
                            }

                            continue
                        }
                    }
                }
            })
        }
    };

    ( $timeout:expr, $( $name:ident: $type:ty => $block:expr ),+ ) => {
        {
            use std::sync::Arc;
            use std::time::{Duration, Instant};
            use std::thread;

            struct Resources {
                $(
                    $name: thread::JoinHandle<Result<$type, ()>>
                ),+
            };

            (|| {
                let timeout = Arc::new($timeout);
                let started_at = Arc::new(Instant::now());

                retry!( @create_multiple_runner timeout, started_at, $( $name: $type => $block ),+ )
            })()
        }
    };

}

#[cfg(test)]
mod test {
    use std::thread;
    use std::sync::{Arc, Mutex};
    use std::time::Duration;

    #[test]
    fn multiple_asynchronous() {
        let completed = Arc::new(Mutex::new(0));

        let c1 = completed.clone();
        let c2 = completed.clone();
        let c3 = completed.clone();

        let num = retry!(Duration::from_secs(5), 
                         num1: i64 => move || {
                             thread::sleep(Duration::from_millis(200));
                             *c1.lock().unwrap() += 1;
                             Ok::<i64, ()>(1)
                         },

                         num2: &'static str => move || {
                             thread::sleep(Duration::from_millis(400));
                             *c2.lock().unwrap() += 1;
                             Ok::<&str, ()>("Asynchronous retries...")
                         },

                         num3: i64 => move || {
                             assert!(*c3.lock().unwrap() == 0);
                             thread::sleep(Duration::from_secs(1));
                             *c3.lock().unwrap() += 1;
                             let num = *c3.lock().unwrap();
                             assert!(num == 3);
                             Ok::<i64, ()>(num)
                         });

        // retry! will return after 1 seconds
        assert!(num == Ok((1, "Asynchronous retries...", 3)));
    }

    #[test]
    fn success_immediately() {
        let fun = || {
            thread::sleep(Duration::from_millis(300));
            Ok::<i64, ()>(23)
        };

        let res = retry!(Duration::from_secs(1), a: i64 => fun);

        assert!(res == Ok(23));
    }

    #[test]
    fn retries_five_times() {
        let retries = Arc::new(Mutex::new(0));
        let fun = {
            let retries = retries.clone();

            move || {
                thread::sleep(Duration::from_millis(100));

                let mut c = retries.lock().unwrap();
                if *c < 5 {
                    *c += 1;
                    return Err(())
                }

                Ok(*c)
            }
        };

        let res = retry!(Duration::from_secs(1), a: i64 => fun);
        assert!(res == Ok(5));
    }

    #[test]
    fn fail_after_timeout() {
        let retries = Arc::new(Mutex::new(0));
        let fun = {
            let retries = retries.clone();

            move || {
                thread::sleep(Duration::from_millis(100));

                let mut c = retries.lock().unwrap();
                *c += 1;

                Err::<i64, ()>(())
            }
        };

        let res = retry!(Duration::from_secs(1), a: i64 => fun);

        assert!(res == Err(()));
        assert!(*retries.lock().unwrap() == 10);
    }

    #[test]
    #[allow(unreachable_code)]
    fn retry_on_panic() {
        let count = Arc::new(Mutex::new(0));

        let c = count.clone();
        let fun = move || {
            *c.lock().unwrap() += 1;

            if *c.lock().unwrap() == 5 {
                return Ok::<i64, ()>(*c.lock().unwrap());
            }

            panic!("Nihahaha!");
        };

        let res = retry!(Duration::from_millis(50), a: i64 => fun);
        assert!(res == Ok(5));
    }
}