smol/examples/hyper-client.rs

//! An HTTP+TLS client based on `hyper` and `async-native-tls`.
//!
//! Run with:
//!
//! ```
//! cargo run --example hyper-client
//! ```

use std::net::Shutdown;
use std::net::{TcpStream, ToSocketAddrs};
use std::pin::Pin;
use std::task::{Context, Poll};

use anyhow::{bail, Context as _, Error, Result};
use async_native_tls::TlsStream;
use http::Uri;
use hyper::{Body, Client, Request, Response};
use smol::{io, prelude::*, Async};

/// Sends a request and fetches the response.
async fn fetch(req: Request<Body>) -> Result<Response<Body>> {
    Ok(Client::builder()
        .executor(SmolExecutor)
        .build::<_, Body>(SmolConnector)
        .request(req)
        .await?)
}

fn main() -> Result<()> {
    smol::block_on(async {
        // Create a request.
        let req = Request::get("https://www.rust-lang.org").body(Body::empty())?;

        // Fetch the response.
        let resp = fetch(req).await?;
        println!("{:#?}", resp);

        // Read the message body.
        let body = resp
            .into_body()
            .try_fold(Vec::new(), |mut body, chunk| {
                body.extend_from_slice(&chunk);
                Ok(body)
            })
            .await?;
        println!("{}", String::from_utf8_lossy(&body));

        Ok(())
    })
}

/// Spawns futures.
#[derive(Clone)]
struct SmolExecutor;

impl<F: Future + Send + 'static> hyper::rt::Executor<F> for SmolExecutor {
    fn execute(&self, fut: F) {
        smol::spawn(async { drop(fut.await) }).detach();
    }
}

/// Connects to URLs.
#[derive(Clone)]
struct SmolConnector;

impl hyper::service::Service<Uri> for SmolConnector {
    type Response = SmolStream;
    type Error = Error;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, uri: Uri) -> Self::Future {
        Box::pin(async move {
            let host = uri.host().context("cannot parse host")?;

            match uri.scheme_str() {
                Some("http") => {
                    let socket_addr = {
                        let host = host.to_string();
                        let port = uri.port_u16().unwrap_or(80);
                        smol::unblock(move || (host.as_str(), port).to_socket_addrs())
                            .await?
                            .next()
                            .context("cannot resolve address")?
                    };
                    let stream = Async::<TcpStream>::connect(socket_addr).await?;
                    Ok(SmolStream::Plain(stream))
                }
                Some("https") => {
                    // In case of HTTPS, establish a secure TLS connection first.
                    let socket_addr = {
                        let host = host.to_string();
                        let port = uri.port_u16().unwrap_or(443);
                        smol::unblock(move || (host.as_str(), port).to_socket_addrs())
                            .await?
                            .next()
                            .context("cannot resolve address")?
                    };
                    let stream = Async::<TcpStream>::connect(socket_addr).await?;
                    let stream = async_native_tls::connect(host, stream).await?;
                    Ok(SmolStream::Tls(stream))
                }
                scheme => bail!("unsupported scheme: {:?}", scheme),
            }
        })
    }
}

/// A TCP or TCP+TLS connection.
enum SmolStream {
    /// A plain TCP connection.
    Plain(Async<TcpStream>),

    /// A TCP connection secured by TLS.
    Tls(TlsStream<Async<TcpStream>>),
}

impl hyper::client::connect::Connection for SmolStream {
    fn connected(&self) -> hyper::client::connect::Connected {
        hyper::client::connect::Connected::new()
    }
}

impl tokio::io::AsyncRead for SmolStream {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        match &mut *self {
            SmolStream::Plain(s) => {
                Pin::new(s)
                    .poll_read(cx, buf.initialize_unfilled())
                    .map_ok(|size| {
                        buf.advance(size);
                    })
            }
            SmolStream::Tls(s) => {
                Pin::new(s)
                    .poll_read(cx, buf.initialize_unfilled())
                    .map_ok(|size| {
                        buf.advance(size);
                    })
            }
        }
    }
}

impl tokio::io::AsyncWrite for SmolStream {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        match &mut *self {
            SmolStream::Plain(s) => Pin::new(s).poll_write(cx, buf),
            SmolStream::Tls(s) => Pin::new(s).poll_write(cx, buf),
        }
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        match &mut *self {
            SmolStream::Plain(s) => Pin::new(s).poll_flush(cx),
            SmolStream::Tls(s) => Pin::new(s).poll_flush(cx),
        }
    }

    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        match &mut *self {
            SmolStream::Plain(s) => {
                s.get_ref().shutdown(Shutdown::Write)?;
                Poll::Ready(Ok(()))
            }
            SmolStream::Tls(s) => Pin::new(s).poll_close(cx),
        }
    }
}
More comments 2020-04-20 15:45:14 +00:00			//! An HTTP+TLS client based on `hyper` and `async-native-tls`.
			`//!`
			`//! Run with:`
			`//!`
			//! ```
			`//! cargo run --example hyper-client`
			//! ```
Lots of examples and docs 2020-03-26 13:57:06 +00:00
Refactor 2020-07-14 19:19:01 +00:00			`use std::net::Shutdown;`
Remove uses of async-net 2020-07-15 09:37:50 +00:00			`use std::net::{TcpStream, ToSocketAddrs};`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`use std::pin::Pin;`
			`use std::task::{Context, Poll};`

			`use anyhow::{bail, Context as _, Error, Result};`
More examples 2020-03-29 15:59:28 +00:00			`use async_native_tls::TlsStream;`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`use http::Uri;`
more examples 2020-03-26 22:36:03 +00:00			`use hyper::{Body, Client, Request, Response};`
Fix compilation errors 2020-08-26 21:59:49 +00:00			`use smol::{io, prelude::*, Async};`
Lots of examples and docs 2020-03-26 13:57:06 +00:00
More comments 2020-04-20 15:45:14 +00:00			`/// Sends a request and fetches the response.`
More examples 2020-03-29 15:59:28 +00:00			`async fn fetch(req: Request<Body>) -> Result<Response<Body>> {`
more examples 2020-03-26 22:36:03 +00:00			`Ok(Client::builder()`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`.executor(SmolExecutor)`
more examples 2020-03-26 22:36:03 +00:00			`.build::<_, Body>(SmolConnector)`
			`.request(req)`
			`.await?)`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`}`

			`fn main() -> Result<()> {`
Fix compilation errors 2020-08-26 21:59:49 +00:00			`smol::block_on(async {`
More comments 2020-04-20 15:45:14 +00:00			`// Create a request.`
more examples 2020-03-26 22:36:03 +00:00			`let req = Request::get("https://www.rust-lang.org").body(Body::empty())?;`
More comments 2020-04-20 15:45:14 +00:00
			`// Fetch the response.`
More examples 2020-03-29 15:59:28 +00:00			`let resp = fetch(req).await?;`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`println!("{:#?}", resp);`

More comments 2020-04-20 15:45:14 +00:00			`// Read the message body.`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`let body = resp`
			`.into_body()`
Reduce the number of dependencies 2020-07-19 22:55:35 +00:00			`.try_fold(Vec::new(), \|mut body, chunk\| {`
more examples 2020-03-26 22:36:03 +00:00			`body.extend_from_slice(&chunk);`
			`Ok(body)`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`})`
			`.await?;`
More and cleaner examples 2020-04-01 19:40:27 +00:00			`println!("{}", String::from_utf8_lossy(&body));`
More comments 2020-04-20 15:45:14 +00:00
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`Ok(())`
			`})`
			`}`

More docs and comments 2020-04-26 17:37:29 +00:00			`/// Spawns futures.`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`#[derive(Clone)]`
			`struct SmolExecutor;`

			`impl<F: Future + Send + 'static> hyper::rt::Executor<F> for SmolExecutor {`
			`fn execute(&self, fut: F) {`
Fix compilation errors 2020-08-26 21:59:49 +00:00			`smol::spawn(async { drop(fut.await) }).detach();`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`}`
			`}`

More docs and comments 2020-04-26 17:37:29 +00:00			`/// Connects to URLs.`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`#[derive(Clone)]`
			`struct SmolConnector;`

			`impl hyper::service::Service<Uri> for SmolConnector {`
			`type Response = SmolStream;`
			`type Error = Error;`
Reduce the number of dependencies 2020-07-19 22:55:35 +00:00			`type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;`
Lots of examples and docs 2020-03-26 13:57:06 +00:00
			`fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {`
			`Poll::Ready(Ok(()))`
			`}`

			`fn call(&mut self, uri: Uri) -> Self::Future {`
			`Box::pin(async move {`
			`let host = uri.host().context("cannot parse host")?;`

			`match uri.scheme_str() {`
			`Some("http") => {`
Remove uses of async-net 2020-07-15 09:37:50 +00:00			`let socket_addr = {`
			`let host = host.to_string();`
			`let port = uri.port_u16().unwrap_or(80);`
Fix compilation errors 2020-08-26 21:59:49 +00:00			`smol::unblock(move \|\| (host.as_str(), port).to_socket_addrs())`
			`.await?`
Remove uses of async-net 2020-07-15 09:37:50 +00:00			`.next()`
			`.context("cannot resolve address")?`
			`};`
			`let stream = Async::<TcpStream>::connect(socket_addr).await?;`
More examples 2020-03-29 15:59:28 +00:00			`Ok(SmolStream::Plain(stream))`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`}`
			`Some("https") => {`
More comments 2020-04-20 15:45:14 +00:00			`// In case of HTTPS, establish a secure TLS connection first.`
Remove uses of async-net 2020-07-15 09:37:50 +00:00			`let socket_addr = {`
			`let host = host.to_string();`
			`let port = uri.port_u16().unwrap_or(443);`
Fix compilation errors 2020-08-26 21:59:49 +00:00			`smol::unblock(move \|\| (host.as_str(), port).to_socket_addrs())`
			`.await?`
Remove uses of async-net 2020-07-15 09:37:50 +00:00			`.next()`
			`.context("cannot resolve address")?`
			`};`
			`let stream = Async::<TcpStream>::connect(socket_addr).await?;`
More examples 2020-03-29 15:59:28 +00:00			`let stream = async_native_tls::connect(host, stream).await?;`
			`Ok(SmolStream::Tls(stream))`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`}`
			`scheme => bail!("unsupported scheme: {:?}", scheme),`
			`}`
			`})`
			`}`
			`}`

More docs and comments 2020-04-26 17:37:29 +00:00			`/// A TCP or TCP+TLS connection.`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`enum SmolStream {`
More docs and comments 2020-04-26 17:37:29 +00:00			`/// A plain TCP connection.`
Remove uses of async-net 2020-07-15 09:37:50 +00:00			`Plain(Async<TcpStream>),`
More docs and comments 2020-04-26 17:37:29 +00:00
			`/// A TCP connection secured by TLS.`
Remove uses of async-net 2020-07-15 09:37:50 +00:00			`Tls(TlsStream<Async<TcpStream>>),`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`}`

			`impl hyper::client::connect::Connection for SmolStream {`
			`fn connected(&self) -> hyper::client::connect::Connected {`
			`hyper::client::connect::Connected::new()`
			`}`
			`}`

			`impl tokio::io::AsyncRead for SmolStream {`
			`fn poll_read(`
			`mut self: Pin<&mut Self>,`
			`cx: &mut Context<'_>,`
Bump dev-dependencies for examples Updates the dependencies used by examples to current - some have had API-breaking changes, so some examples are updated to catch up with those as well. Note that the hyper-server example doesn't seem to work when tested, but it also didn't work before this PR. Looked at it a bit and found that poll_accept() returns Pending via incoming.poll_next() and then poll_accept() isn't invoked again. hyper-client meanwhile works fine, so the issue doesn't seem to be related to the AsyncRead updates. 2021-01-04 10:40:26 +00:00			`buf: &mut tokio::io::ReadBuf<'_>,`
			`) -> Poll<io::Result<()>> {`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`match &mut *self {`
Bump dev-dependencies for examples Updates the dependencies used by examples to current - some have had API-breaking changes, so some examples are updated to catch up with those as well. Note that the hyper-server example doesn't seem to work when tested, but it also didn't work before this PR. Looked at it a bit and found that poll_accept() returns Pending via incoming.poll_next() and then poll_accept() isn't invoked again. hyper-client meanwhile works fine, so the issue doesn't seem to be related to the AsyncRead updates. 2021-01-04 10:40:26 +00:00			`SmolStream::Plain(s) => {`
			`Pin::new(s)`
			`.poll_read(cx, buf.initialize_unfilled())`
			`.map_ok(\|size\| {`
			`buf.advance(size);`
			`})`
			`}`
			`SmolStream::Tls(s) => {`
			`Pin::new(s)`
			`.poll_read(cx, buf.initialize_unfilled())`
			`.map_ok(\|size\| {`
			`buf.advance(size);`
			`})`
			`}`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`}`
			`}`
			`}`

			`impl tokio::io::AsyncWrite for SmolStream {`
			`fn poll_write(`
			`mut self: Pin<&mut Self>,`
			`cx: &mut Context<'_>,`
			`buf: &[u8],`
			`) -> Poll<io::Result<usize>> {`
			`match &mut *self {`
More examples 2020-03-29 15:59:28 +00:00			`SmolStream::Plain(s) => Pin::new(s).poll_write(cx, buf),`
			`SmolStream::Tls(s) => Pin::new(s).poll_write(cx, buf),`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`}`
			`}`

more examples 2020-03-26 22:36:03 +00:00			`fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`match &mut *self {`
More examples 2020-03-29 15:59:28 +00:00			`SmolStream::Plain(s) => Pin::new(s).poll_flush(cx),`
			`SmolStream::Tls(s) => Pin::new(s).poll_flush(cx),`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`}`
			`}`

Lots of changes 2020-03-31 16:04:16 +00:00			`fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`match &mut *self {`
Lots of changes 2020-03-31 16:04:16 +00:00			`SmolStream::Plain(s) => {`
Remove uses of async-net 2020-07-15 09:37:50 +00:00			`s.get_ref().shutdown(Shutdown::Write)?;`
Lots of changes 2020-03-31 16:04:16 +00:00			`Poll::Ready(Ok(()))`
			`}`
			`SmolStream::Tls(s) => Pin::new(s).poll_close(cx),`
Lots of examples and docs 2020-03-26 13:57:06 +00:00			`}`
			`}`
			`}`