-
-
Notifications
You must be signed in to change notification settings - Fork 50
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Remove
Clone constraints and buffer the service (#348)
* feat: remove the `Clone` requirements for services * test save * fix: get buffered layer working * update: remove clone & update api * fix: tests and api * lint: clippy fixes * lint: cargo fmt
- Loading branch information
1 parent
dd9570c
commit 5861e84
Showing
17 changed files
with
613 additions
and
71 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| //! Error types for the `Buffer` middleware. | ||
| use std::{fmt, sync::Arc}; | ||
| use tower::BoxError; | ||
|
|
||
| /// An error produced by a [`Service`] wrapped by a [`Buffer`] | ||
| /// | ||
| /// [`Service`]: crate::Service | ||
| /// [`Buffer`]: crate::buffer::Buffer | ||
| #[derive(Debug)] | ||
| pub(crate) struct ServiceError { | ||
| inner: Arc<BoxError>, | ||
| } | ||
|
|
||
| /// An error produced when the a buffer's worker closes unexpectedly. | ||
| pub(crate) struct Closed { | ||
| _p: (), | ||
| } | ||
|
|
||
| // ===== impl ServiceError ===== | ||
|
|
||
| impl ServiceError { | ||
| pub(crate) fn new(inner: BoxError) -> ServiceError { | ||
| let inner = Arc::new(inner); | ||
| ServiceError { inner } | ||
| } | ||
|
|
||
| // Private to avoid exposing `Clone` trait as part of the public API | ||
| pub(crate) fn clone(&self) -> ServiceError { | ||
| ServiceError { | ||
| inner: self.inner.clone(), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl fmt::Display for ServiceError { | ||
| fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { | ||
| write!(fmt, "buffered service failed: {}", self.inner) | ||
| } | ||
| } | ||
|
|
||
| impl std::error::Error for ServiceError { | ||
| fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { | ||
| Some(&**self.inner) | ||
| } | ||
| } | ||
|
|
||
| // ===== impl Closed ===== | ||
|
|
||
| impl Closed { | ||
| pub(crate) fn new() -> Self { | ||
| Closed { _p: () } | ||
| } | ||
| } | ||
|
|
||
| impl fmt::Debug for Closed { | ||
| fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { | ||
| fmt.debug_tuple("Closed").finish() | ||
| } | ||
| } | ||
|
|
||
| impl fmt::Display for Closed { | ||
| fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { | ||
| fmt.write_str("buffer's worker closed unexpectedly") | ||
| } | ||
| } | ||
|
|
||
| impl std::error::Error for Closed {} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| //! Future types for the [`Buffer`] middleware. | ||
| //! | ||
| //! [`Buffer`]: crate::buffer::Buffer | ||
| use super::{error::Closed, message}; | ||
| use futures::ready; | ||
| use pin_project_lite::pin_project; | ||
| use std::{ | ||
| future::Future, | ||
| pin::Pin, | ||
| task::{Context, Poll}, | ||
| }; | ||
|
|
||
| pin_project! { | ||
| /// Future that completes when the buffered service eventually services the submitted request. | ||
| #[derive(Debug)] | ||
| pub struct ResponseFuture<T> { | ||
| #[pin] | ||
| state: ResponseState<T>, | ||
| } | ||
| } | ||
|
|
||
| pin_project! { | ||
| #[project = ResponseStateProj] | ||
| #[derive(Debug)] | ||
| enum ResponseState<T> { | ||
| Failed { | ||
| error: Option<tower::BoxError>, | ||
| }, | ||
| Rx { | ||
| #[pin] | ||
| rx: message::Rx<T>, | ||
| }, | ||
| Poll { | ||
| #[pin] | ||
| fut: T, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| impl<T> ResponseFuture<T> { | ||
| pub(crate) fn new(rx: message::Rx<T>) -> Self { | ||
| ResponseFuture { | ||
| state: ResponseState::Rx { rx }, | ||
| } | ||
| } | ||
|
|
||
| pub(crate) fn failed(err: tower::BoxError) -> Self { | ||
| ResponseFuture { | ||
| state: ResponseState::Failed { error: Some(err) }, | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl<F, T, E> Future for ResponseFuture<F> | ||
| where | ||
| F: Future<Output = Result<T, E>>, | ||
| E: Into<tower::BoxError>, | ||
| { | ||
| type Output = Result<T, tower::BoxError>; | ||
|
|
||
| fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { | ||
| let mut this = self.project(); | ||
|
|
||
| loop { | ||
| match this.state.as_mut().project() { | ||
| ResponseStateProj::Failed { error } => { | ||
| return Poll::Ready(Err(error.take().expect("polled after error"))); | ||
| } | ||
| ResponseStateProj::Rx { rx } => match ready!(rx.poll(cx)) { | ||
| Ok(Ok(fut)) => this.state.set(ResponseState::Poll { fut }), | ||
| Ok(Err(e)) => return Poll::Ready(Err(e.into())), | ||
| Err(_) => return Poll::Ready(Err(Closed::new().into())), | ||
| }, | ||
| ResponseStateProj::Poll { fut } => return fut.poll(cx).map_err(Into::into), | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| use futures::channel::oneshot; | ||
|
|
||
| use super::error::ServiceError; | ||
|
|
||
| /// Message sent over buffer | ||
| #[derive(Debug)] | ||
| pub(crate) struct Message<Request, Fut> { | ||
| pub(crate) request: Request, | ||
| pub(crate) tx: Tx<Fut>, | ||
| } | ||
|
|
||
| /// Response sender | ||
| pub(crate) type Tx<Fut> = oneshot::Sender<Result<Fut, ServiceError>>; | ||
|
|
||
| /// Response receiver | ||
| pub(crate) type Rx<Fut> = oneshot::Receiver<Result<Fut, ServiceError>>; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| pub(crate) mod error; | ||
| pub(crate) mod future; | ||
| pub(crate) mod message; | ||
| pub(crate) mod service; | ||
| pub(crate) mod worker; |
Oops, something went wrong.