feat(service): use tower_service::Service for hyper::service

hyperium · Aug 20, 2019 · ec520d5 · ec520d5
1 parent 53a437c
commit ec520d5
Show file tree

Hide file tree

Showing 11 changed files with 230 additions and 111 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -38,6 +38,7 @@ pin-utils = "0.1.0-alpha.4"
 time = "0.1"
 tokio = { version = "0.2.0-alpha.2", optional = true, default-features = false, features = ["rt-full"] }
 #tokio-buf = "0.2.0-alpha.1"
+tower-service = "=0.3.0-alpha.1"
 tokio-executor = "0.2.0-alpha.2"
 tokio-io = "0.2.0-alpha.2"
 tokio-sync = "0.2.0-alpha.2"

diff --git a/examples/tower_server.rs b/examples/tower_server.rs
@@ -0,0 +1,70 @@
+#![feature(async_await)]
+#![deny(warnings)]
+
+use hyper::{Body, Request, Response, Server};
+use tower_service::Service;
+use futures_util::future;
+use std::task::{Context, Poll};
+
+const ROOT: &'static str = "/";
+
+#[derive(Debug)]
+pub struct Svc;
+
+impl Service<Request<Body>> for Svc {
+    type Response = Response<Body>;
+    type Error = hyper::Error;
+    type Future = future::Ready<Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Ok(()).into()
+    }
+
+    fn call(&mut self, req: Request<Body>) -> Self::Future {
+        let mut rsp = Response::builder();
+
+        let uri = req.uri();
+        if uri.path() != ROOT {
+            let body = Body::from(Vec::new());
+            let rsp = rsp.status(404).body(body).unwrap();
+            return future::ok(rsp);
+        }
+
+        let body = Body::from(Vec::from(&b"heyo!"[..]));
+        let rsp = rsp.status(200).body(body).unwrap();
+        future::ok(rsp)
+    }
+}
+
+pub struct MakeSvc;
+
+impl<T> Service<T> for MakeSvc {
+    type Response = Svc;
+    type Error = std::io::Error;
+    type Future = future::Ready<Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Ok(()).into()
+    }
+
+    fn call(&mut self, _: T) -> Self::Future {
+        future::ok(Svc)
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    pretty_env_logger::init();
+
+    let addr = "127.0.0.1:1337".parse().unwrap();
+
+
+    let server = Server::bind(&addr)
+        .serve(MakeSvc);
+
+    println!("Listening on http://{}", addr);
+
+    server.await?;
+
+    Ok(())
+}
diff --git a/src/common/exec.rs b/src/common/exec.rs
@@ -5,7 +5,7 @@ use std::sync::Arc;
 
 use tokio_executor::{SpawnError, TypedExecutor};
 
-use crate::body::Payload;
+use crate::body::{Payload, Body};
 use crate::proto::h2::server::H2Stream;
 use crate::server::conn::spawn_all::{NewSvcTask, Watcher};
 use crate::service::Service;
@@ -14,7 +14,7 @@ pub trait H2Exec<F, B: Payload>: Clone {
     fn execute_h2stream(&mut self, fut: H2Stream<F, B>) -> crate::Result<()>;
 }
 
-pub trait NewSvcExec<I, N, S: Service, E, W: Watcher<I, S, E>>: Clone {
+pub trait NewSvcExec<I, N, S: Service<Body>, E, W: Watcher<I, S, E>>: Clone {
     fn execute_new_svc(&mut self, fut: NewSvcTask<I, N, S, E, W>) -> crate::Result<()>;
 }
 
@@ -119,7 +119,7 @@ where
 impl<I, N, S, E, W> NewSvcExec<I, N, S, E, W> for Exec
 where
     NewSvcTask<I, N, S, E, W>: Future<Output=()> + Send + 'static,
-    S: Service,
+    S: Service<Body>,
     W: Watcher<I, S, E>,
 {
     fn execute_new_svc(&mut self, fut: NewSvcTask<I, N, S, E, W>) -> crate::Result<()> {
@@ -148,7 +148,7 @@ impl<I, N, S, E, W> NewSvcExec<I, N, S, E, W> for E
 where
     E: TypedExecutor<NewSvcTask<I, N, S, E, W>> + Clone,
     NewSvcTask<I, N, S, E, W>: Future<Output=()>,
-    S: Service,
+    S: Service<Body>,
     W: Watcher<I, S, E>,
 {
     fn execute_new_svc(&mut self, fut: NewSvcTask<I, N, S, E, W>) -> crate::Result<()> {

diff --git a/src/proto/h1/dispatch.rs b/src/proto/h1/dispatch.rs
@@ -30,7 +30,7 @@ pub(crate) trait Dispatch {
     fn should_poll(&self) -> bool;
 }
 
-pub struct Server<S: Service> {
+pub struct Server<S: Service<B>, B> {
     in_flight: Pin<Box<Option<S::Future>>>,
     pub(crate) service: S,
 }
@@ -412,11 +412,11 @@ impl<'a, T> Drop for OptGuard<'a, T> {
 
 // ===== impl Server =====
 
-impl<S> Server<S>
+impl<S, B> Server<S, B>
 where
-    S: Service,
+    S: Service<B>,
 {
-    pub fn new(service: S) -> Server<S> {
+    pub fn new(service: S) -> Server<S, B> {
         Server {
             in_flight: Box::pin(None),
             service: service,
@@ -429,11 +429,11 @@ where
 }
 
 // Service is never pinned
-impl<S: Service> Unpin for Server<S> {}
+impl<S: Service<B>, B> Unpin for Server<S, B> {}
 
-impl<S, Bs> Dispatch for Server<S>
+impl<S, Bs> Dispatch for Server<S, Body>
 where
-    S: Service<ReqBody=Body, ResBody=Bs>,
+    S: Service<Body, ResBody=Bs>,
     S::Error: Into<Box<dyn StdError + Send + Sync>>,
     Bs: Payload,
 {

diff --git a/src/proto/h2/server.rs b/src/proto/h2/server.rs
@@ -20,7 +20,7 @@ use crate::{Body, Response};
 
 pub(crate) struct Server<T, S, B, E>
 where
-    S: Service,
+    S: Service<Body>,
     B: Payload,
 {
     exec: E,
@@ -29,7 +29,7 @@ where
 }
 
 // TODO: fix me
-impl<T, S: Service, B: Payload, E> Unpin for Server<T, S, B, E> {}
+impl<T, S: Service<Body>, B: Payload, E> Unpin for Server<T, S, B, E> {}
 
 enum State<T, B>
 where
@@ -52,7 +52,7 @@ where
 impl<T, S, B, E> Server<T, S, B, E>
 where
     T: AsyncRead + AsyncWrite + Unpin,
-    S: Service<ReqBody=Body, ResBody=B>,
+    S: Service<Body, ResBody=B>,
     S::Error: Into<Box<dyn StdError + Send + Sync>>,
     B: Payload,
     B::Data: Unpin,
@@ -90,7 +90,7 @@ where
 impl<T, S, B, E> Future for Server<T, S, B, E>
 where
     T: AsyncRead + AsyncWrite + Unpin,
-    S: Service<ReqBody=Body, ResBody=B>,
+    S: Service<Body, ResBody=B>,
     S::Error: Into<Box<dyn StdError + Send + Sync>>,
     B: Payload,
     B::Data: Unpin,
@@ -133,7 +133,7 @@ where
     fn poll_server<S, E>(&mut self, cx: &mut task::Context<'_>, service: &mut S, exec: &mut E) -> Poll<crate::Result<()>>
     where
         S: Service<
-            ReqBody=Body,
+            Body,
             ResBody=B,
         >,
         S::Error: Into<Box<dyn StdError + Send + Sync>>,