wip

Author: seanmonstar

src/client/mod.rs

@@ -265,7 +265,7 @@ where T: AsyncRead + AsyncWrite + 'static,
     type Response = proto::ResponseHead;
     type ResponseBody = proto::Chunk;
     type Error = ::Error;
-    type Transport = proto::Conn<T, B::Item, proto::ClientTransaction, Pooled<TokioClient<B>>>;
+    type Transport = proto::Conn<T, B::Item, proto::ClientTransaction, proto::ProtoOnion, Pooled<TokioClient<B>>>;
     type BindTransport = BindingClient<T, B>;
 
     fn bind_transport(&self, io: T) -> Self::BindTransport {
@@ -286,13 +286,13 @@ where T: AsyncRead + AsyncWrite + 'static,
       B: Stream<Error=::Error>,
       B::Item: AsRef<[u8]>,
 {
-    type Item = proto::Conn<T, B::Item, proto::ClientTransaction, Pooled<TokioClient<B>>>;
+    type Item = proto::Conn<T, B::Item, proto::ClientTransaction, proto::ProtoOnion, Pooled<TokioClient<B>>>;
     type Error = io::Error;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         match self.rx.poll() {
             Ok(Async::Ready(client)) => Ok(Async::Ready(
-                    proto::Conn::new(self.io.take().expect("binding client io lost"), client)
+                    proto::Conn::new(self.io.take().expect("binding client io lost"), client, proto::ProtoOnion)
             )),
             Ok(Async::NotReady) => Ok(Async::NotReady),
             Err(_canceled) => unreachable!(),

src/proto/conn.rs

@@ -2,7 +2,7 @@ use std::fmt;
 use std::io::{self, Write};
 use std::marker::PhantomData;
 
-use futures::{Poll, Async, AsyncSink, Stream, Sink, StartSend};
+use futures::{Poll, Async, AsyncSink, Future, Stream, Sink, StartSend};
 use futures::task::Task;
 use tokio_io::{AsyncRead, AsyncWrite};
 use tokio_proto::streaming::pipeline::{Frame, Transport};
@@ -21,21 +21,23 @@ use version::HttpVersion;
 /// The connection will determine when a message begins and ends as well as
 /// determine if this  connection can be kept alive after the message,
 /// or if it is complete.
-pub struct Conn<I, B, T, K = KA> {
+pub struct Conn<I, B, T, O, K = KA> {
     io: Buffered<I>,
+    onion: O,
     state: State<B, K>,
     _marker: PhantomData<T>
 }
 
-impl<I, B, T, K> Conn<I, B, T, K>
+impl<I, B, T, O, K> Conn<I, B, T, O, K>
 where I: AsyncRead + AsyncWrite,
       B: AsRef<[u8]>,
       T: Http1Transaction,
       K: KeepAlive
 {
-    pub fn new(io: I, keep_alive: K) -> Conn<I, B, T, K> {
+    pub fn new(io: I, keep_alive: K, onion: O) -> Conn<I, B, T, O, K> {
         Conn {
             io: Buffered::new(io),
+            onion: onion,
             state: State {
                 keep_alive: keep_alive,
                 method: None,
@@ -51,8 +53,8 @@ where I: AsyncRead + AsyncWrite,
         self.io.set_flush_pipeline(enabled);
     }
 
-    fn poll2(&mut self) -> Poll<Option<Frame<super::MessageHead<T::Incoming>, super::Chunk, ::Error>>, io::Error> {
-        trace!("Conn::poll()");
+    fn poll_incoming(&mut self) -> Poll<Option<Frame<super::MessageHead<T::Incoming>, super::Chunk, ::Error>>, io::Error> {
+        trace!("Conn::poll_incoming()");
 
         loop {
             if self.is_read_closed() {
@@ -412,7 +414,30 @@ where I: AsyncRead + AsyncWrite,
     }
 }
 
-impl<I, B, T, K> Stream for Conn<I, B, T, K>
+// ==== future impl ====
+
+pub struct NoOnion;
+
+impl<I, B, T, K> Future for Conn<I, B, T, NoOnion, K>
+where I: AsyncRead + AsyncWrite,
+      B: AsRef<[u8]>,
+      T: Http1Transaction,
+      K: KeepAlive,
+      T::Outgoing: fmt::Debug {
+    type Item = I;
+    type Error = ::Error;
+
+    #[inline]
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        unimplemented!("future impl")
+    }
+}
+
+// ==== tokio_proto impl ====
+
+pub struct ProtoOnion;
+
+impl<I, B, T, K> Stream for Conn<I, B, T, ProtoOnion, K>
 where I: AsyncRead + AsyncWrite,
       B: AsRef<[u8]>,
       T: Http1Transaction,
@@ -423,14 +448,14 @@ where I: AsyncRead + AsyncWrite,
 
     #[inline]
     fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-        self.poll2().map_err(|err| {
+        self.poll_incoming().map_err(|err| {
             debug!("poll error: {}", err);
             err
         })
     }
 }
 
-impl<I, B, T, K> Sink for Conn<I, B, T, K>
+impl<I, B, T, K> Sink for Conn<I, B, T, ProtoOnion, K>
 where I: AsyncRead + AsyncWrite,
       B: AsRef<[u8]>,
       T: Http1Transaction,
@@ -501,14 +526,14 @@ where I: AsyncRead + AsyncWrite,
     }
 }
 
-impl<I, B, T, K> Transport for Conn<I, B, T, K>
+impl<I, B, T, K> Transport for Conn<I, B, T, ProtoOnion, K>
 where I: AsyncRead + AsyncWrite + 'static,
       B: AsRef<[u8]> + 'static,
       T: Http1Transaction + 'static,
       K: KeepAlive + 'static,
       T::Outgoing: fmt::Debug {}
 
-impl<I, B: AsRef<[u8]>, T, K: fmt::Debug> fmt::Debug for Conn<I, B, T, K> {
+impl<I, B: AsRef<[u8]>, T, O, K: fmt::Debug> fmt::Debug for Conn<I, B, T, O, K> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         f.debug_struct("Conn")
             .field("state", &self.state)

src/proto/mod.rs

@@ -12,7 +12,7 @@ use uri::Uri;
 use version::HttpVersion;
 use version::HttpVersion::{Http10, Http11};
 
-pub use self::conn::{Conn, KeepAlive, KA};
+pub use self::conn::{Conn, KeepAlive, KA, NoOnion, ProtoOnion};
 pub use self::body::{Body, TokioBody};
 pub use self::chunk::Chunk;
 

src/server/mod.rs

@@ -165,6 +165,18 @@ impl<B: AsRef<[u8]> + 'static> Http<B> {
         })
     }
 
+    pub fn no_onion<S, I, Bd>(&self, io: I, _service: S) -> Connection<I, B, S>
+        where S: Service<Request = Request, Response = Response<Bd>, Error = ::Error> + 'static,
+              Bd: Stream<Item=B, Error=::Error> + 'static,
+              I: AsyncRead + AsyncWrite + 'static,
+
+    {
+        Connection {
+            conn: proto::Conn::new(io, proto::KA::Busy, proto::NoOnion),
+            _marker: PhantomData,
+        }
+    }
+
     /// Bind a `Service` using types from the `http` crate.
     ///
     /// See `Http::bind_connection`.
@@ -185,6 +197,11 @@ impl<B: AsRef<[u8]> + 'static> Http<B> {
     }
 }
 
+pub struct Connection<I, B, S> {
+    conn: proto::Conn<I, B, proto::ServerTransaction, proto::NoOnion>,
+    _marker: PhantomData<S>,
+}
+
 impl<B> Clone for Http<B> {
     fn clone(&self) -> Http<B> {
         Http {
@@ -210,11 +227,11 @@ pub struct __ProtoRequest(proto::RequestHead);
 pub struct __ProtoResponse(ResponseHead);
 #[doc(hidden)]
 #[allow(missing_debug_implementations)]
-pub struct __ProtoTransport<T, B>(proto::Conn<T, B, proto::ServerTransaction>);
+pub struct __ProtoTransport<T, B>(proto::Conn<T, B, proto::ServerTransaction, proto::ProtoOnion>);
 #[doc(hidden)]
 #[allow(missing_debug_implementations)]
 pub struct __ProtoBindTransport<T, B> {
-    inner: future::FutureResult<proto::Conn<T, B, proto::ServerTransaction>, io::Error>,
+    inner: future::FutureResult<proto::Conn<T, B, proto::ServerTransaction, proto::ProtoOnion>, io::Error>,
 }
 
 impl<T, B> ServerProto<T> for Http<B>
@@ -236,7 +253,7 @@ impl<T, B> ServerProto<T> for Http<B>
         } else {
             proto::KA::Disabled
         };
-        let mut conn = proto::Conn::new(io, ka);
+        let mut conn = proto::Conn::new(io, ka, proto::ProtoOnion);
         conn.set_flush_pipeline(self.pipeline);
         __ProtoBindTransport {
             inner: future::ok(conn),