beekeeper 0.3.0

A full-featured worker pool library for parallelizing tasks
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159

//! Error types that may be returned by `Worker`s.
use crate::panic::Panic;
use std::fmt::Debug;

/// Error that can result from applying a `Worker`'s function to an input.
#[derive(thiserror::Error, Debug)]
pub enum ApplyError<I, E> {
    /// The task failed due to a fatal error that cannot be retried.
    #[error("Task failed (and is not retryable)")]
    Fatal { input: Option<I>, error: E },
    /// The task failed due to a (possibly) transient error and can be retried.
    #[error("Task failed, but is retryable")]
    Retryable { input: I, error: E },
    /// The task was cancelled before it completed.
    #[error("Task was cancelled")]
    Cancelled { input: I },
    /// The task panicked.
    #[error("Task panicked")]
    Panic {
        input: Option<I>,
        payload: Panic<String>,
    },
}

impl<I, E> ApplyError<I, E> {
    /// Returns the input value associated with this error, if available.
    pub fn input(&self) -> Option<&I> {
        match self {
            Self::Fatal { input, .. } => input.as_ref(),
            Self::Retryable { input, .. } => Some(input),
            Self::Cancelled { input, .. } => Some(input),
            Self::Panic { input, .. } => input.as_ref(),
        }
    }

    /// Consumes this `ApplyError` and returns the input value associated with it, if available.
    pub fn into_input(self) -> Option<I> {
        match self {
            Self::Fatal { input, .. } => input,
            Self::Retryable { input, .. } => Some(input),
            Self::Cancelled { input, .. } => Some(input),
            Self::Panic { input, .. } => input,
        }
    }

    /// Consumes this `ApplyError` and:
    /// * Panics, if this is a `Panic` variant,
    /// * Returns `None`, if this is a `Cancelled` variant,
    /// * Returns `Some(E)` otherwise
    pub fn into_source(self) -> Option<E> {
        match self {
            Self::Fatal { input: _, error } => Some(error),
            Self::Retryable { input: _, error } => Some(error),
            Self::Cancelled { .. } => None,
            Self::Panic { input: _, payload } => payload.resume(),
        }
    }
}

/// Error that can result from applying a `RefWorker`'s function to an input.
#[derive(thiserror::Error, Debug)]
pub enum ApplyRefError<E> {
    /// The task failed due to a fatal error that cannot be retried.
    #[error("Error is not retryable")]
    Fatal(E),
    /// The task failed due to a (possibly) transient error and can be retried.
    #[error("Error is retryable")]
    Retryable(E),
    /// The task was cancelled before it completed.
    #[error("Task was cancelled")]
    Cancelled,
}

impl<E> ApplyRefError<E> {
    pub(super) fn into_apply_error<I>(self, input: I) -> ApplyError<I, E> {
        match self {
            Self::Fatal(error) => ApplyError::Fatal {
                input: Some(input),
                error,
            },
            Self::Retryable(error) => ApplyError::Retryable { input, error },
            Self::Cancelled => ApplyError::Cancelled { input },
        }
    }
}

impl<E> From<E> for ApplyRefError<E> {
    fn from(e: E) -> Self {
        ApplyRefError::Fatal(e)
    }
}

#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests {
    use super::ApplyError;
    use crate::panic::Panic;

    type TestError<'a> = ApplyError<usize, &'a str>;

    impl<I, E> ApplyError<I, E> {
        pub fn panic(input: Option<I>, detail: Option<String>) -> Self {
            Self::Panic {
                input,
                payload: Panic::new("test", detail),
            }
        }
    }

    #[test]
    fn test_input() {
        let cancelled: TestError = ApplyError::Cancelled { input: 42 };
        assert_eq!(&42, cancelled.input().unwrap());
        assert_eq!(42, cancelled.into_input().unwrap());

        let retryable: TestError = ApplyError::Retryable {
            input: 42,
            error: "bork",
        };
        assert_eq!(&42, retryable.input().unwrap());
        assert_eq!(42, retryable.into_input().unwrap());

        let not_retryable: TestError = ApplyError::Fatal {
            input: Some(42),
            error: "bork",
        };
        assert_eq!(&42, not_retryable.input().unwrap());
        assert_eq!(42, not_retryable.into_input().unwrap());

        let panic: TestError = ApplyError::panic(Some(42), None);
        assert_eq!(&42, panic.input().unwrap());
        assert_eq!(42, panic.into_input().unwrap());
    }

    #[test]
    fn test_error() {
        let cancelled: TestError = ApplyError::Cancelled { input: 42 };
        assert_eq!(None, cancelled.into_source());

        let retryable: TestError = ApplyError::Retryable {
            input: 42,
            error: "bork",
        };
        assert_eq!(Some("bork"), retryable.into_source());

        let not_retryable: TestError = ApplyError::Fatal {
            input: Some(42),
            error: "bork",
        };
        assert_eq!(Some("bork"), not_retryable.into_source());
    }

    #[test]
    #[should_panic]
    fn test_panic() {
        let panic: TestError = ApplyError::panic(Some(42), Some("borked".to_string()));
        let _ = panic.into_source();
    }
}