simple_hyper_client/connector/
http.rs

1/* Copyright (c) Fortanix, Inc.
2 *
3 * This Source Code Form is subject to the terms of the Mozilla Public
4 * License, v. 2.0. If a copy of the MPL was not distributed with this
5 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
6
7use crate::connector::{NetworkConnection, NetworkConnector};
8use crate::connector_impl::connect;
9use hyper::Uri;
10use hyper_util::client::legacy::connect::{Connected, Connection};
11use std::error::Error as StdError;
12use std::future::Future;
13use std::pin::Pin;
14use std::task::{Context, Poll};
15use std::time::Duration;
16use std::{fmt, io};
17use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
18use tokio::net::TcpStream;
19
20pub(crate) const DEFAULT_HTTP_PORT: u16 = 80;
21pub(crate) const DEFAULT_HTTPS_PORT: u16 = 443;
22
23/// A simple HTTP connector
24///
25/// NOTE: this provides less functionality than [hyper's `HttpConnector`].
26///
27/// [hyper's `HttpConnector`]: https://docs.rs/hyper/0.14/hyper/client/struct.HttpConnector.html
28#[derive(Clone)]
29pub struct HttpConnector {
30    connect_timeout: Option<Duration>,
31}
32
33impl HttpConnector {
34    pub fn new() -> Self {
35        HttpConnector {
36            connect_timeout: None,
37        }
38    }
39
40    /// Set the connect timeout. Default is None.
41    pub fn connect_timeout(mut self, timeout: Option<Duration>) -> Self {
42        self.connect_timeout = timeout;
43        self
44    }
45
46    /// Connects to an endpoint and returns a [`TcpStream`] for the connection.
47    pub async fn connect_raw(
48        &self,
49        uri: Uri,
50    ) -> Result<TcpStream, Box<dyn StdError + Send + Sync>> {
51        connect(uri, false, self.connect_timeout)
52            .await
53            .map(|conn| conn.into_tcp_stream())
54            .map_err(|err| Box::new(err) as _)
55    }
56}
57
58impl NetworkConnector for HttpConnector {
59    fn connect(
60        &self,
61        uri: Uri,
62    ) -> Pin<
63        Box<dyn Future<Output = Result<NetworkConnection, Box<dyn StdError + Send + Sync>>> + Send>,
64    > {
65        let connect_timeout = self.connect_timeout;
66        Box::pin(async move {
67            match connect(uri, false, connect_timeout).await {
68                Ok(conn) => Ok(NetworkConnection::new(conn)),
69                Err(e) => Err(Box::new(e) as _),
70            }
71        })
72    }
73}
74
75/// A wrapper around [`tokio::net::TcpStream`]
76///
77/// [`tokio::net::TcpStream`]: https://docs.rs/tokio/1.0/tokio/net/struct.TcpStream.html
78pub struct HttpConnection {
79    pub(crate) stream: TcpStream,
80}
81
82impl Connection for HttpConnection {
83    fn connected(&self) -> Connected {
84        // TODO(#13): provide remote address
85        Connected::new()
86    }
87}
88
89impl HttpConnection {
90    pub fn into_tcp_stream(self) -> TcpStream {
91        self.stream
92    }
93}
94
95impl AsyncRead for HttpConnection {
96    fn poll_read(
97        self: Pin<&mut Self>,
98        cx: &mut Context<'_>,
99        buf: &mut ReadBuf<'_>,
100    ) -> Poll<io::Result<()>> {
101        Pin::new(&mut self.get_mut().stream).poll_read(cx, buf)
102    }
103}
104
105impl AsyncWrite for HttpConnection {
106    fn poll_write(
107        self: Pin<&mut Self>,
108        cx: &mut Context<'_>,
109        buf: &[u8],
110    ) -> Poll<io::Result<usize>> {
111        Pin::new(&mut self.get_mut().stream).poll_write(cx, buf)
112    }
113
114    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
115        Pin::new(&mut self.get_mut().stream).poll_flush(cx)
116    }
117
118    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
119        Pin::new(&mut self.get_mut().stream).poll_shutdown(cx)
120    }
121}
122
123pub struct ConnectError {
124    msg: &'static str,
125    cause: Option<Box<dyn StdError + Send + Sync>>,
126}
127
128impl ConnectError {
129    pub fn new(msg: &'static str) -> Self {
130        ConnectError { msg, cause: None }
131    }
132
133    pub fn cause<E: Into<Box<dyn StdError + Send + Sync>>>(mut self, cause: E) -> Self {
134        self.cause = Some(cause.into());
135        self
136    }
137}
138
139impl fmt::Debug for ConnectError {
140    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
141        if let Some(ref cause) = self.cause {
142            f.debug_tuple("ConnectError")
143                .field(&self.msg)
144                .field(cause)
145                .finish()
146        } else {
147            self.msg.fmt(f)
148        }
149    }
150}
151
152impl fmt::Display for ConnectError {
153    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
154        f.write_str(&self.msg)?;
155        if let Some(ref cause) = self.cause {
156            write!(f, ": {}", cause)?;
157        }
158        Ok(())
159    }
160}
161
162impl StdError for ConnectError {
163    fn source(&self) -> Option<&(dyn StdError + 'static)> {
164        self.cause.as_ref().map(|e| &**e as _)
165    }
166}
simple_hyper_client/connector/http.rs

simple_hyper_client/connector/
http.rs
