apalis_core/backend/ext/
inspect_err.rs

1use std::{
2    pin::Pin,
3    task::{Context, Poll},
4};
5
6use futures_sink::Sink;
7use futures_util::SinkExt;
8use futures_util::TryStreamExt;
9
10use crate::{backend::*, task::Task, worker::context::WorkerContext};
11
12/// A `Backend` wrapper that runs a callback `F` on each error yielded by the poll stream.
13#[derive(Debug, Clone)]
14pub struct InspectErr<B, F> {
15    pub(super) backend: B,
16    pub(super) f: F,
17}
18
19impl<B, F> Backend for InspectErr<B, F>
20where
21    B: Backend,
22    F: FnMut(&B::Error),
23{
24    type Task = B::Task;
25    type Error = B::Error;
26
27    fn poll_ready(
28        &mut self,
29        cx: &mut Context<'_>,
30        worker: &WorkerContext,
31    ) -> Poll<Result<(), Self::Error>> {
32        self.backend.poll_ready(cx, worker).map_err(|err| {
33            (self.f)(&err);
34            err
35        })
36    }
37
38    fn poll_next(
39        &mut self,
40        cx: &mut Context<'_>,
41        worker: &WorkerContext,
42    ) -> Poll<Option<Result<Self::Task, Self::Error>>> {
43        self.backend.poll_next(cx, worker).map(|opt| {
44            opt.map(|res| {
45                res.inspect_err(|err| {
46                    (self.f)(err);
47                })
48            })
49        })
50    }
51
52    fn poll_close(
53        &mut self,
54        cx: &mut Context<'_>,
55        worker: &WorkerContext,
56    ) -> Poll<Result<(), Self::Error>> {
57        self.backend.poll_close(cx, worker).map_err(|err| {
58            (self.f)(&err);
59            err
60        })
61    }
62}
63
64impl<B, F, T, Err> Sink<T> for InspectErr<B, F>
65where
66    B: Sink<T, Error = Err> + Unpin,
67    F: FnMut(&Err) + Unpin,
68{
69    type Error = Err;
70    fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
71        let this = self.get_mut();
72        this.backend.start_send_unpin(item).inspect_err(|err| {
73            (this.f)(err);
74        })
75    }
76    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
77        let this = self.get_mut();
78        this.backend.poll_ready_unpin(cx).map_err(|err| {
79            (this.f)(&err);
80            err
81        })
82    }
83    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
84        let this = self.get_mut();
85        this.backend.poll_flush_unpin(cx).map_err(|err| {
86            (this.f)(&err);
87            err
88        })
89    }
90    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
91        let this = self.get_mut();
92        this.backend.poll_close_unpin(cx).map_err(|err| {
93            (this.f)(&err);
94            err
95        })
96    }
97}
98
99delegate_config!(InspectErr<B, F>, backend);
100
101delegate_codec!(InspectErr<B, F>, backend);
102
103delegate_deref!(InspectErr<B, F>, backend);
104
105delegate_expose!(
106    impl<B, F> for InspectErr<B, F>
107    where {
108            B: Backend + Send,
109            F: FnMut(&B::Error) + Send + Sync + Clone + 'static,
110    }
111    => backend,
112    wrap = |this, result| { let mut f = this.f.clone(); result.inspect_err(move |err| {(f)(err)}) }
113);
apalis_core/backend/ext/inspect_err.rs

apalis_core/backend/ext/
inspect_err.rs
