Allow errors (#26)

This changes error model to actually allow errors. I think if we're going to use this for things like tonic's route we need a more flexible error handling model. The same `handle_error` adaptors are still there but services aren't required to have `Infallible` as their error type. The error type is simply propagated all the way through.

src/buffer.rs

@@ -0,0 +1,190 @@
+use futures_util::ready;
+use pin_project::pin_project;
+use std::{
+    future::Future,
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+};
+use tokio::sync::{mpsc, oneshot, OwnedSemaphorePermit, Semaphore};
+use tokio_util::sync::PollSemaphore;
+use tower::{Service, ServiceExt};
+
+/// A version of [`tower::buffer::Buffer`] which panicks on channel related errors, thus keeping
+/// the error type of the service.
+pub(crate) struct MpscBuffer<S, R>
+where
+    S: Service<R>,
+{
+    tx: mpsc::UnboundedSender<Msg<S, R>>,
+    semaphore: PollSemaphore,
+    permit: Option<OwnedSemaphorePermit>,
+}
+
+impl<S, R> Clone for MpscBuffer<S, R>
+where
+    S: Service<R>,
+{
+    fn clone(&self) -> Self {
+        Self {
+            tx: self.tx.clone(),
+            semaphore: self.semaphore.clone(),
+            permit: None,
+        }
+    }
+}
+
+impl<S, R> MpscBuffer<S, R>
+where
+    S: Service<R>,
+{
+    pub(crate) fn new(svc: S) -> Self
+    where
+        S: Send + 'static,
+        R: Send + 'static,
+        S::Error: Send + 'static,
+        S::Future: Send + 'static,
+    {
+        let (tx, rx) = mpsc::unbounded_channel::<Msg<S, R>>();
+        let semaphore = PollSemaphore::new(Arc::new(Semaphore::new(1024)));
+
+        tokio::spawn(run_worker(svc, rx));
+
+        Self {
+            tx,
+            semaphore,
+            permit: None,
+        }
+    }
+}
+
+async fn run_worker<S, R>(mut svc: S, mut rx: mpsc::UnboundedReceiver<Msg<S, R>>)
+where
+    S: Service<R>,
+{
+    while let Some((req, reply_tx)) = rx.recv().await {
+        match svc.ready().await {
+            Ok(svc) => {
+                let future = svc.call(req);
+                let _ = reply_tx.send(WorkerReply::Future(future));
+            }
+            Err(err) => {
+                let _ = reply_tx.send(WorkerReply::Error(err));
+            }
+        }
+    }
+}
+
+type Msg<S, R> = (
+    R,
+    oneshot::Sender<WorkerReply<<S as Service<R>>::Future, <S as Service<R>>::Error>>,
+);
+
+enum WorkerReply<F, E> {
+    Future(F),
+    Error(E),
+}
+
+impl<S, R> Service<R> for MpscBuffer<S, R>
+where
+    S: Service<R>,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = ResponseFuture<S::Future, S::Error>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        if self.permit.is_some() {
+            return Poll::Ready(Ok(()));
+        }
+
+        let permit = ready!(self.semaphore.poll_acquire(cx))
+            .expect("buffer semaphore closed. This is a bug in awebframework and should never happen. Please file an issue");
+
+        self.permit = Some(permit);
+
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: R) -> Self::Future {
+        let permit = self
+            .permit
+            .take()
+            .expect("semaphore permit missing. Did you forget to call `poll_ready`?");
+
+        let (reply_tx, reply_rx) = oneshot::channel::<WorkerReply<S::Future, S::Error>>();
+
+        self.tx.send((req, reply_tx)).unwrap_or_else(|_| {
+            panic!("buffer worker not running. This is a bug in awebframework and should never happen. Please file an issue")
+        });
+
+        ResponseFuture {
+            state: State::Channel(reply_rx),
+            permit,
+        }
+    }
+}
+
+#[pin_project]
+pub(crate) struct ResponseFuture<F, E> {
+    #[pin]
+    state: State<F, E>,
+    permit: OwnedSemaphorePermit,
+}
+
+#[pin_project(project = StateProj)]
+enum State<F, E> {
+    Channel(oneshot::Receiver<WorkerReply<F, E>>),
+    Future(#[pin] F),
+}
+
+impl<F, E, T> Future for ResponseFuture<F, E>
+where
+    F: Future<Output = Result<T, E>>,
+{
+    type Output = Result<T, E>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        loop {
+            let mut this = self.as_mut().project();
+
+            let new_state = match this.state.as_mut().project() {
+                StateProj::Channel(reply_rx) => {
+                    let msg = ready!(Pin::new(reply_rx).poll(cx))
+                        .expect("buffer worker not running. This is a bug in awebframework and should never happen. Please file an issue");
+
+                    match msg {
+                        WorkerReply::Future(future) => State::Future(future),
+                        WorkerReply::Error(err) => return Poll::Ready(Err(err)),
+                    }
+                }
+                StateProj::Future(future) => {
+                    return future.poll(cx);
+                }
+            };
+
+            this.state.set(new_state);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[allow(unused_imports)]
+    use super::*;
+    use tower::ServiceExt;
+
+    #[tokio::test]
+    async fn test_buffer() {
+        let mut svc = MpscBuffer::new(tower::service_fn(handle));
+
+        let res = svc.ready().await.unwrap().call(42).await.unwrap();
+
+        assert_eq!(res, "foo");
+    }
+
+    async fn handle(req: i32) -> Result<&'static str, std::convert::Infallible> {
+        assert_eq!(req, 42);
+        Ok("foo")
+    }
+}