apalis_core/backend/ext/
map_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 maps the backend's error type `Self::Error` into `E2`.
13#[derive(Debug, Clone)]
14pub struct MapErr<B, F> {
15    pub(super) backend: B,
16    pub(super) f: F,
17}
18
19impl<B, F, E2> Backend for MapErr<B, F>
20where
21    B: Backend,
22    F: Fn(B::Error) -> E2,
23    E2: std::error::Error + Send + Sync + 'static,
24{
25    type Task = B::Task;
26    type Error = E2;
27
28    fn poll_ready(
29        &mut self,
30        cx: &mut Context<'_>,
31        worker: &WorkerContext,
32    ) -> Poll<Result<(), Self::Error>> {
33        self.backend
34            .poll_ready(cx, worker)
35            .map_err(|err| (self.f)(err))
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
44            .poll_next(cx, worker)
45            .map(|opt| opt.map(|res| res.map_err(|err| (self.f)(err))))
46    }
47
48    fn poll_close(
49        &mut self,
50        cx: &mut Context<'_>,
51        worker: &WorkerContext,
52    ) -> Poll<Result<(), Self::Error>> {
53        self.backend
54            .poll_close(cx, worker)
55            .map_err(|err| (self.f)(err))
56    }
57}
58
59impl<B, F, T, Err, E2> Sink<T> for MapErr<B, F>
60where
61    B: Sink<T, Error = Err> + Unpin,
62    F: Fn(B::Error) -> E2 + Unpin,
63    E2: std::error::Error + Send + Sync + 'static,
64{
65    type Error = E2;
66    fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
67        let this = self.get_mut();
68        this.backend
69            .start_send_unpin(item)
70            .map_err(|err| (this.f)(err))
71    }
72    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
73        let this = self.get_mut();
74        this.backend
75            .poll_ready_unpin(cx)
76            .map_err(|err| (this.f)(err))
77    }
78    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
79        let this = self.get_mut();
80        this.backend
81            .poll_flush_unpin(cx)
82            .map_err(|err| (this.f)(err))
83    }
84    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
85        let this = self.get_mut();
86        this.backend
87            .poll_close_unpin(cx)
88            .map_err(|err| (this.f)(err))
89    }
90}
91
92delegate_deref!(MapErr<B, F>, backend);
93
94delegate_config!(MapErr<B, F>, backend);
95
96delegate_codec!(MapErr<B, F>, backend);
97
98delegate_expose!(
99    impl<B, F, E2> for MapErr<B, F>
100    where {
101            B: Backend + Send,
102            F: Fn(B::Error) -> E2 + Send + Sync + Clone + 'static,
103            E2: std::error::Error + Send + Sync + 'static,
104    }
105    => backend,
106    wrap = |this, result| {
107        let f = this.f.clone();
108        #[allow(clippy::manual_inspect, clippy::redundant_closure)]
109        result.map_err(move |err| (f)(err))
110    }
111);
apalis_core/backend/ext/map_err.rs

apalis_core/backend/ext/
map_err.rs
