tower_resilience_executor/
service.rs

1//! Service implementation for the executor middleware.
2
3use crate::Executor;
4use pin_project_lite::pin_project;
5use std::future::Future;
6use std::pin::Pin;
7use std::task::{Context, Poll};
8use tokio::sync::oneshot;
9use tower_service::Service;
10
11/// A service that delegates request processing to an executor.
12///
13/// Each request is spawned as a new task on the executor, allowing
14/// parallel processing of multiple requests.
15///
16/// # Requirements
17///
18/// The inner service must implement `Clone` so that each spawned task
19/// can have its own instance. This is the standard pattern for Tower
20/// services that need to be shared across tasks.
21///
22/// # Cancellation
23///
24/// When the response future is dropped, the spawned task continues
25/// to run to completion. This is intentional to avoid partial processing.
26/// If you need cancellation, consider wrapping with a timeout layer.
27#[derive(Clone)]
28pub struct ExecutorService<S, E> {
29    inner: S,
30    executor: E,
31}
32
33impl<S, E> ExecutorService<S, E> {
34    /// Creates a new executor service.
35    pub fn new(service: S, executor: E) -> Self {
36        Self {
37            inner: service,
38            executor,
39        }
40    }
41
42    /// Returns a reference to the inner service.
43    pub fn get_ref(&self) -> &S {
44        &self.inner
45    }
46
47    /// Returns a mutable reference to the inner service.
48    pub fn get_mut(&mut self) -> &mut S {
49        &mut self.inner
50    }
51
52    /// Consumes the service and returns the inner service.
53    pub fn into_inner(self) -> S {
54        self.inner
55    }
56}
57
58impl<S, E, Req> Service<Req> for ExecutorService<S, E>
59where
60    S: Service<Req> + Clone + Send + 'static,
61    S::Future: Send,
62    S::Response: Send + 'static,
63    S::Error: Send + 'static,
64    E: Executor,
65    Req: Send + 'static,
66{
67    type Response = S::Response;
68    type Error = ExecutorError<S::Error>;
69    type Future = ExecutorFuture<S::Response, S::Error>;
70
71    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
72        // Poll the inner service for readiness
73        self.inner.poll_ready(cx).map_err(ExecutorError::Service)
74    }
75
76    fn call(&mut self, req: Req) -> Self::Future {
77        // Take the readied service for the spawned task, leaving a fresh
78        // clone behind for the next poll_ready cycle. See #286.
79        let clone = self.inner.clone();
80        let mut service = std::mem::replace(&mut self.inner, clone);
81        let (tx, rx) = oneshot::channel();
82
83        // Spawn the request processing on the executor
84        let _handle = self.executor.spawn(async move {
85            // Call the service
86            let result = service.call(req).await;
87
88            // Send the result back
89            // The send may fail if the receiver is dropped (caller cancelled)
90            // We ignore this error since there's nothing useful to do.
91            let _ = tx.send(result.map_err(ExecutorError::Service));
92        });
93
94        ExecutorFuture { rx }
95    }
96}
97
98/// Error type for executor service operations.
99#[derive(Debug, Clone, PartialEq, Eq)]
100pub enum ExecutorError<E> {
101    /// The spawned task was cancelled or panicked.
102    TaskCancelled,
103    /// The inner service returned an error.
104    Service(E),
105}
106
107impl<E: std::fmt::Display> std::fmt::Display for ExecutorError<E> {
108    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
109        match self {
110            Self::TaskCancelled => write!(f, "executor task was cancelled"),
111            Self::Service(e) => write!(f, "service error: {}", e),
112        }
113    }
114}
115
116impl<E: std::error::Error + 'static> std::error::Error for ExecutorError<E> {
117    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
118        match self {
119            Self::Service(e) => Some(e),
120            _ => None,
121        }
122    }
123}
124
125pin_project! {
126    /// Future returned by [`ExecutorService`].
127    pub struct ExecutorFuture<T, E> {
128        #[pin]
129        rx: oneshot::Receiver<Result<T, ExecutorError<E>>>,
130    }
131}
132
133impl<T, E> Future for ExecutorFuture<T, E> {
134    type Output = Result<T, ExecutorError<E>>;
135
136    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
137        let this = self.project();
138        match this.rx.poll(cx) {
139            Poll::Ready(Ok(result)) => Poll::Ready(result),
140            Poll::Ready(Err(_)) => Poll::Ready(Err(ExecutorError::TaskCancelled)),
141            Poll::Pending => Poll::Pending,
142        }
143    }
144}
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149
150    #[test]
151    fn test_error_display() {
152        let err: ExecutorError<std::io::Error> = ExecutorError::TaskCancelled;
153        assert_eq!(err.to_string(), "executor task was cancelled");
154    }
155
156    #[test]
157    fn test_error_eq() {
158        let err1: ExecutorError<&str> = ExecutorError::TaskCancelled;
159        let err2: ExecutorError<&str> = ExecutorError::TaskCancelled;
160        assert_eq!(err1, err2);
161
162        let err3: ExecutorError<&str> = ExecutorError::Service("test");
163        let err4: ExecutorError<&str> = ExecutorError::Service("test");
164        assert_eq!(err3, err4);
165    }
166}
tower_resilience_executor/service.rs

tower_resilience_executor/
service.rs
