tower/buffer/
layer.rs

1use super::service::Buffer;
2use std::{fmt, marker::PhantomData};
3use tower_layer::Layer;
4use tower_service::Service;
5
6/// Adds an mpsc buffer in front of an inner service.
7///
8/// The default Tokio executor is used to run the given service,
9/// which means that this layer can only be used on the Tokio runtime.
10///
11/// See the module documentation for more details.
12pub struct BufferLayer<Request> {
13    bound: usize,
14    _p: PhantomData<fn(Request)>,
15}
16
17impl<Request> BufferLayer<Request> {
18    /// Creates a new [`BufferLayer`] with the provided `bound`.
19    ///
20    /// `bound` gives the maximal number of requests that can be queued for the service before
21    /// backpressure is applied to callers.
22    ///
23    ///
24    /// # Panics
25    ///
26    /// Panics if `bound` is zero.
27    ///
28    ///
29    /// # A note on choosing a `bound`
30    ///
31    /// When [`Buffer`]'s implementation of [`poll_ready`] returns [`Poll::Ready`], it reserves a
32    /// slot in the channel for the forthcoming [`call`]. However, if this call doesn't arrive,
33    /// this reserved slot may be held up for a long time. As a result, it's advisable to set
34    /// `bound` to be at least the maximum number of concurrent requests the [`Buffer`] will see.
35    /// If you do not, all the slots in the buffer may be held up by futures that have just called
36    /// [`poll_ready`] but will not issue a [`call`], which prevents other senders from issuing new
37    /// requests.
38    ///
39    /// [`Poll::Ready`]: std::task::Poll::Ready
40    /// [`call`]: crate::Service::call
41    /// [`poll_ready`]: crate::Service::poll_ready
42    pub const fn new(bound: usize) -> Self {
43        assert!(bound > 0, "buffer bound must be greater than zero");
44        BufferLayer {
45            bound,
46            _p: PhantomData,
47        }
48    }
49}
50
51impl<S, Request> Layer<S> for BufferLayer<Request>
52where
53    S: Service<Request> + Send + 'static,
54    S::Future: Send,
55    S::Error: Into<crate::BoxError> + Send + Sync,
56    Request: Send + 'static,
57{
58    type Service = Buffer<Request, S::Future>;
59
60    fn layer(&self, service: S) -> Self::Service {
61        Buffer::new(service, self.bound)
62    }
63}
64
65impl<Request> fmt::Debug for BufferLayer<Request> {
66    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
67        f.debug_struct("BufferLayer")
68            .field("bound", &self.bound)
69            .finish()
70    }
71}
72
73impl<Request> Clone for BufferLayer<Request> {
74    fn clone(&self) -> Self {
75        *self
76    }
77}
78
79impl<Request> Copy for BufferLayer<Request> {}
tower/buffer/layer.rs

tower/buffer/
layer.rs
