init upgrade submodule

Author: yoshuawuyts

src/lib.rs

@@ -131,11 +131,14 @@ mod status_code;
 mod version;
 
 cfg_unstable! {
+    pub mod upgrade;
+
     mod client;
     mod server;
 
     pub use client::Client;
     pub use server::Server;
+
 }
 
 pub use body::Body;

src/request.rs

@@ -16,6 +16,10 @@ use crate::mime::Mime;
 use crate::trailers::{self, Trailers};
 use crate::{Body, Extensions, Method, StatusCode, Url, Version};
 
+cfg_unstable! {
+    use crate::upgrade;
+}
+
 pin_project_lite::pin_project! {
     /// An HTTP request.
     ///
@@ -33,8 +37,10 @@ pin_project_lite::pin_project! {
         url: Url,
         headers: Headers,
         version: Option<Version>,
-        sender: Option<sync::Sender<Trailers>>,
-        receiver: Option<sync::Receiver<Trailers>>,
+        trailers_sender: Option<sync::Sender<Trailers>>,
+        trailers_receiver: Option<sync::Receiver<Trailers>>,
+        upgrade_sender: Option<sync::Sender<upgrade::Connection>>,
+        upgrade_receiver: Option<sync::Receiver<upgrade::Connection>>,
         #[pin]
         body: Body,
         local_addr: Option<String>,
@@ -45,24 +51,51 @@ pin_project_lite::pin_project! {
 
 impl Request {
     /// Create a new request.
+    #[cfg(feature = "unstable")]
     pub fn new<U>(method: Method, url: U) -> Self
     where
         U: TryInto<Url>,
         U::Error: std::fmt::Debug,
     {
         let url = url.try_into().expect("Could not convert into a valid url");
-        let (sender, receiver) = sync::channel(1);
+        let (trailers_sender, trailers_receiver) = sync::channel(1);
+        let (upgrade_sender, upgrade_receiver) = sync::channel(1);
         Self {
             method,
             url,
             headers: Headers::new(),
             version: None,
             body: Body::empty(),
-            sender: Some(sender),
-            receiver: Some(receiver),
             ext: Extensions::new(),
             peer_addr: None,
             local_addr: None,
+            trailers_receiver: Some(trailers_receiver),
+            trailers_sender: Some(trailers_sender),
+            upgrade_sender: Some(upgrade_sender),
+            upgrade_receiver: Some(upgrade_receiver),
+        }
+    }
+
+    /// Create a new request.
+    #[cfg(not(feature = "unstable"))]
+    pub fn new<U>(method: Method, url: U) -> Self
+    where
+        U: TryInto<Url>,
+        U::Error: std::fmt::Debug,
+    {
+        let url = url.try_into().expect("Could not convert into a valid url");
+        let (trailers_sender, trailers_receiver) = sync::channel(1);
+        Self {
+            method,
+            url,
+            headers: Headers::new(),
+            version: None,
+            body: Body::empty(),
+            ext: Extensions::new(),
+            peer_addr: None,
+            local_addr: None,
+            trailers_receiver: Some(trailers_receiver),
+            trailers_sender: Some(trailers_sender),
         }
     }
 
@@ -543,7 +576,7 @@ impl Request {
     /// Sends trailers to the a receiver.
     pub fn send_trailers(&mut self) -> trailers::Sender {
         let sender = self
-            .sender
+            .trailers_sender
             .take()
             .expect("Trailers sender can only be constructed once");
         trailers::Sender::new(sender)
@@ -552,12 +585,34 @@ impl Request {
     /// Receive trailers from a sender.
     pub async fn recv_trailers(&mut self) -> trailers::Receiver {
         let receiver = self
-            .receiver
+            .trailers_receiver
             .take()
             .expect("Trailers receiver can only be constructed once");
         trailers::Receiver::new(receiver)
     }
 
+    /// Sends an upgrade connection to the a receiver.
+    #[cfg(feature = "unstable")]
+    #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+    pub fn send_upgrade(&mut self) -> upgrade::Sender {
+        let sender = self
+            .upgrade_sender
+            .take()
+            .expect("Upgrade sender can only be constructed once");
+        upgrade::Sender::new(sender)
+    }
+
+    /// Receive an upgraded connection from a sender.
+    #[cfg(feature = "unstable")]
+    #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+    pub async fn recv_upgrade(&mut self) -> upgrade::Receiver {
+        let receiver = self
+            .upgrade_receiver
+            .take()
+            .expect("Upgrade receiver can only be constructed once");
+        upgrade::Receiver::new(receiver)
+    }
+
     /// An iterator visiting all header pairs in arbitrary order.
     pub fn iter(&self) -> headers::Iter<'_> {
         self.headers.iter()
@@ -867,8 +922,10 @@ impl Clone for Request {
             url: self.url.clone(),
             headers: self.headers.clone(),
             version: self.version.clone(),
-            sender: self.sender.clone(),
-            receiver: self.receiver.clone(),
+            trailers_sender: None,
+            trailers_receiver: None,
+            upgrade_sender: None,
+            upgrade_receiver: None,
             body: Body::empty(),
             ext: Extensions::new(),
             peer_addr: self.peer_addr.clone(),

src/upgrade/connection.rs

@@ -0,0 +1,42 @@
+use async_std::io::{self, prelude::*};
+
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// An upgraded HTTP connection.
+#[derive(Debug, Clone)]
+pub struct Connection {
+    inner: Pin<Box<dyn Read + Write + Clone + Send + Sync + Unpin + 'static>>,
+}
+
+impl Read for Connection {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        let this = self.project();
+        Pin::new(this.inner).poll_read(cx, buf)
+    }
+}
+
+impl Write for Connection {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        let this = self.project();
+        Pin::new(this.inner).poll_write(cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let this = self.project();
+        Pin::new(this.inner).poll_flush(cx)
+    }
+
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let this = self.project();
+        Pin::new(this.inner).poll_close(cx)
+    }
+}

src/upgrade/mod.rs

@@ -0,0 +1,19 @@
+//! HTTP protocol upgrades.
+//!
+//! In HTTP it's not uncommon to convert from one protocol to another. For
+//! example `HTTP/1.1` can upgrade a connection to websockets using the
+//! [upgrade header](https://developer.mozilla.org/en-US/docs/Web/HTTP/Protocol_upgrade_mechanism),
+//! while `HTTP/2` uses [a custom
+//! handshake](https://tools.ietf.org/html/rfc8441#section-5.1). Regardless of
+//! the HTTP version, changing protocols always involves some handshake,
+//! after which it is turned into a stream of bytes. This module provides
+//! primitives for upgrading from HTTP request-response pairs to alternate
+//! protocols.
+
+mod connection;
+mod receiver;
+mod sender;
+
+pub use connection::Connection;
+pub use receiver::Receiver;
+pub use sender::Sender;

src/upgrade/receiver.rs

@@ -0,0 +1,35 @@
+use async_std::prelude::*;
+use async_std::sync;
+
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use crate::upgrade::Connection;
+
+/// The receiving half of a channel to send an upgraded connection.
+///
+/// Unlike `async_std::sync::channel` the `send` method on this type can only be
+/// called once, and cannot be cloned. That's because only a single instance of
+/// `Connection` should be created.
+#[must_use = "Futures do nothing unless polled or .awaited"]
+#[derive(Debug)]
+pub struct Receiver {
+    receiver: sync::Receiver<Connection>,
+}
+
+impl Receiver {
+    /// Create a new instance of `Receiver`.
+    #[allow(unused)]
+    pub(crate) fn new(receiver: sync::Receiver<Connection>) -> Self {
+        Self { receiver }
+    }
+}
+
+impl Future for Receiver {
+    type Output = Option<Connection>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut self.receiver).poll_next(cx)
+    }
+}

src/upgrade/sender.rs

@@ -0,0 +1,28 @@
+use async_std::sync;
+
+use crate::upgrade::Connection;
+
+/// The sending half of a channel to send an upgraded connection.
+///
+/// Unlike `async_std::sync::channel` the `send` method on this type can only be
+/// called once, and cannot be cloned. That's because only a single instance of
+/// `Connection` should be created.
+#[derive(Debug)]
+pub struct Sender {
+    sender: sync::Sender<Connection>,
+}
+
+impl Sender {
+    /// Create a new instance of `Sender`.
+    #[doc(hidden)]
+    pub fn new(sender: sync::Sender<Connection>) -> Self {
+        Self { sender }
+    }
+
+    /// Send a `Trailer`.
+    ///
+    /// The channel will be consumed after having sent trailers.
+    pub async fn send(self, trailers: Connection) {
+        self.sender.send(trailers).await
+    }
+}