feat(client): add support for upgrade request

client/src/client.rs

@@ -20,6 +20,7 @@ use crate::{
     service::HttpService,
     timeout::{Timeout, TimeoutConfig},
     tls::connector::Connector,
+    upgrade::UpgradeRequest,
     uri::Uri,
 };
 
@@ -184,6 +185,64 @@ impl Client {
         self.request_builder(url, Method::CONNECT).mutate_marker()
     }
 
+    #[cfg(feature = "http1")]
+    /// Start a new upgrade request.
+    ///
+    /// # Example
+    /// ```rust
+    /// use xitca_client::{Client, bytes::Bytes, http::Method};
+    ///
+    /// async fn _main() -> Result<(), xitca_client::error::Error> {
+    /// // construct a new client and initialize connect request.
+    /// let client = Client::new();
+    /// let mut upgrade_response = client
+    ///     .upgrade("http://localhost:8080", Method::GET)
+    ///     .protocol("protocol1, protocol2")
+    ///     .send().await?
+    /// ;
+    ///
+    /// if let Some(upgrade) = upgrade_response.headers.get(xitca_client::http::header::UPGRADE) {
+    ///    // check which protocol it was upgraded to
+    /// }
+    ///
+    /// // upgrade_response is a response that contains the http request head and tunnel connection.
+    ///
+    /// // import Stream trait and call it's method on tunnel to receive bytes.
+    /// use futures::StreamExt;
+    /// if let Some(Ok(_)) = upgrade_response.tunnel().next().await {
+    ///     // received bytes data.
+    /// }
+    ///
+    /// // import Sink trait and call it's method on tunnel to send bytes data.
+    /// use futures::SinkExt;
+    /// // send bytes data.
+    /// upgrade_response.tunnel().send(b"996").await?;
+    ///
+    /// // tunnel support split sending/receiving task into different parts to enable concurrent bytes data handling.
+    /// let (_head, mut tunnel) = upgrade_response.into_parts();
+    /// let (mut write, mut read) = tunnel.split();
+    ///
+    /// // read part can operate with Stream trait implement.
+    /// if let Some(Ok(_)) = read.next().await {
+    ///     // received bytes data.
+    /// }
+    ///
+    /// // write part can operate with Sink trait implement.
+    /// write.send(b"996").await?;
+    ///
+    /// Ok(())
+    /// # }
+    /// ```
+    pub fn upgrade<U>(&self, url: U, method: Method) -> UpgradeRequest<'_>
+    where
+        uri::Uri: TryFrom<U>,
+        Error: From<<uri::Uri as TryFrom<U>>::Error>,
+    {
+        self.request_builder(url, method)
+            .version(Version::HTTP_11)
+            .mutate_marker()
+    }
+
     #[cfg(all(feature = "websocket", feature = "http1"))]
     /// Start a new websocket request.
     ///

client/src/http_tunnel.rs

@@ -51,13 +51,13 @@ impl HttpTunnelRequest<'_> {
 }
 
 pub struct HttpTunnel {
-    body: ResponseBody,
-    io: TunnelIo,
+    pub(crate) body: ResponseBody,
+    pub(crate) io: TunnelIo,
 }
 
 // io type to bridge AsyncIo trait and h2 body's poll based read/write apis.
 #[derive(Default)]
-struct TunnelIo {
+pub(crate) struct TunnelIo {
     write_buf: BytesMut,
     #[cfg(feature = "http2")]
     adaptor: TunnelIoAdaptor,

client/src/lib.rs

@@ -43,6 +43,7 @@ mod service;
 mod timeout;
 mod tls;
 mod tunnel;
+mod upgrade;
 mod uri;
 
 #[cfg(feature = "http1")]

client/src/request.rs

@@ -19,7 +19,7 @@ use crate::{
 /// builder type for [http::Request] with extended functionalities.
 pub struct RequestBuilder<'a, M = marker::Http> {
     pub(crate) req: http::Request<BoxBody>,
-    err: Vec<Error>,
+    pub(crate) err: Vec<Error>,
     client: &'a Client,
     timeout: Duration,
     _marker: PhantomData<M>,

client/src/upgrade.rs

@@ -0,0 +1,102 @@
+//! http upgrade handling.
+
+use super::{
+    error::{Error, ErrorResponse},
+    http::{
+        header::{self, HeaderValue},
+        response::Parts,
+        StatusCode,
+    },
+    http_tunnel::HttpTunnel,
+    request::RequestBuilder,
+    tunnel::Tunnel,
+};
+use std::ops::{Deref, DerefMut};
+
+pub type UpgradeRequest<'a> = RequestBuilder<'a, marker::Upgrade>;
+pub type UpgradeRequestWithProtocol<'a> = RequestBuilder<'a, marker::UpgradeWithProtocol>;
+
+mod marker {
+    pub struct Upgrade;
+    pub struct UpgradeWithProtocol;
+}
+
+pub struct UpgradeResponse {
+    pub parts: Parts,
+    pub tunnel: Tunnel<HttpTunnel>,
+}
+
+impl<'a> UpgradeRequest<'a> {
+    pub fn protocol<V>(mut self, proto: V) -> UpgradeRequestWithProtocol<'a>
+    where
+        V: TryInto<HeaderValue>,
+        <V as TryInto<HeaderValue>>::Error: std::error::Error + Send + Sync + 'static,
+    {
+        match proto.try_into() {
+            Ok(v) => {
+                self.req.headers_mut().insert(header::UPGRADE, v);
+            }
+            Err(e) => {
+                self.push_error(Error::Std(Box::new(e)));
+            }
+        };
+
+        self.mutate_marker()
+    }
+}
+
+impl UpgradeRequestWithProtocol<'_> {
+    /// Send the request and wait for response asynchronously.
+    pub async fn send(mut self) -> Result<UpgradeResponse, Error> {
+        self.headers_mut()
+            .insert(header::CONNECTION, HeaderValue::from_static("upgrade"));
+
+        let res = self._send().await?;
+
+        let status = res.status();
+        let expect_status = StatusCode::SWITCHING_PROTOCOLS;
+        if status != expect_status {
+            return Err(Error::from(ErrorResponse {
+                expect_status,
+                status,
+                description: "upgrade tunnel can't be established",
+            }));
+        }
+
+        let (parts, body) = res.into_inner().into_parts();
+
+        Ok(UpgradeResponse {
+            parts,
+            tunnel: Tunnel::new(HttpTunnel {
+                body,
+                io: Default::default(),
+            }),
+        })
+    }
+}
+
+impl UpgradeResponse {
+    #[inline]
+    pub fn into_parts(self) -> (Parts, Tunnel<HttpTunnel>) {
+        (self.parts, self.tunnel)
+    }
+
+    #[inline]
+    pub fn tunnel(&mut self) -> &mut Tunnel<HttpTunnel> {
+        &mut self.tunnel
+    }
+}
+
+impl Deref for UpgradeResponse {
+    type Target = Parts;
+
+    fn deref(&self) -> &Self::Target {
+        &self.parts
+    }
+}
+
+impl DerefMut for UpgradeResponse {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.parts
+    }
+}