Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Upgrade hyper examples to v1.0 and use smol-hyper #296

Merged
merged 2 commits into from
Jan 16, 2024
Merged

Upgrade hyper examples to v1.0 and use smol-hyper #296

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5dc6544
ex: Use smol-hyper in hyper-client example
notgull Jan 15, 2024
76b4686
ex: Use smol-hyper in hyper-server example
notgull Jan 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,11 +37,15 @@ ctrlc = "3"
doc-comment = "0.3"
futures = "0.3"
http = "0.2"
http-body-util = "0.1.0"
http-types = "2"
hyper = { version = "0.14", default-features = false, features = ["client", "http1", "server", "stream"] }
hyper = { version = "1.0", default-features = false, features = ["client", "http1", "server"] }
macro_rules_attribute = "0.2.0"
native-tls = "0.2"
scraper = "0.18"
signal-hook = "0.3"
smol-hyper = "0.1.0"
smol-macros = "0.1.0"
surf = { version = "2", default-features = false, features = ["h1-client"] }
tempfile = "3"
tokio = { version = "1", default-features = false, features = ["rt-multi-thread"] }
Expand Down
214 changes: 89 additions & 125 deletions examples/hyper-client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,175 +6,139 @@
//! cargo run --example hyper-client
//! ```

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

use anyhow::{bail, Context as _, Error, Result};
use anyhow::{bail, Context as _, Result};
use async_native_tls::TlsStream;
use http::Uri;
use hyper::{Body, Client, Request, Response};
use smol::{io, prelude::*, Async};
use http_body_util::{BodyStream, Empty};
use hyper::body::Incoming;
use hyper::{Request, Response};
use macro_rules_attribute::apply;
use smol::{io, net::TcpStream, prelude::*, Executor};
use smol_hyper::rt::FuturesIo;
use smol_macros::main;

/// 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?)
}
async fn fetch(
ex: &Executor<'static>,
req: Request<Empty<&'static [u8]>>,
) -> Result<Response<Incoming>> {
// Connect to the HTTP server.
let io = {
let host = req.uri().host().context("cannot parse host")?;

match req.uri().scheme_str() {
Some("http") => {
let stream = {
let port = req.uri().port_u16().unwrap_or(80);
TcpStream::connect((host, port)).await?
};
SmolStream::Plain(stream)
}
Some("https") => {
// In case of HTTPS, establish a secure TLS connection first.
let stream = {
let port = req.uri().port_u16().unwrap_or(443);
TcpStream::connect((host, port)).await?
};
let stream = async_native_tls::connect(host, stream).await?;
SmolStream::Tls(stream)
}
scheme => bail!("unsupported scheme: {:?}", scheme),
}
};

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(())
// Spawn the HTTP/1 connection.
let (mut sender, conn) = hyper::client::conn::http1::handshake(FuturesIo::new(io)).await?;
ex.spawn(async move {
if let Err(e) = conn.await {
println!("Connection failed: {:?}", e);
}
})
}
.detach();

/// 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();
}
// Get the result
let result = sender.send_request(req).await?;
Ok(result)
}

/// 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),
#[apply(main!)]
async fn main(ex: &Executor<'static>) -> Result<()> {
// Create a request.
let url: hyper::Uri = "https://www.rust-lang.org".try_into()?;
let req = Request::builder()
.header(
hyper::header::HOST,
url.authority().unwrap().clone().as_str(),
)
.uri(url)
.body(Empty::new())?;

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

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

Ok(())
}

/// A TCP or TCP+TLS connection.
enum SmolStream {
/// A plain TCP connection.
Plain(Async<TcpStream>),
Plain(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()
}
Tls(TlsStream<TcpStream>),
}

impl tokio::io::AsyncRead for SmolStream {
impl AsyncRead for SmolStream {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<io::Result<()>> {
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
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);
})
}
SmolStream::Plain(stream) => Pin::new(stream).poll_read(cx, buf),
SmolStream::Tls(stream) => Pin::new(stream).poll_read(cx, buf),
}
}
}

impl tokio::io::AsyncWrite for SmolStream {
impl 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),
SmolStream::Plain(stream) => Pin::new(stream).poll_write(cx, buf),
SmolStream::Tls(stream) => Pin::new(stream).poll_write(cx, buf),
}
}

fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
fn poll_close(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),
SmolStream::Plain(stream) => Pin::new(stream).poll_close(cx),
SmolStream::Tls(stream) => Pin::new(stream).poll_close(cx),
}
}

fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
fn poll_flush(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),
SmolStream::Plain(stream) => Pin::new(stream).poll_flush(cx),
SmolStream::Tls(stream) => Pin::new(stream).poll_flush(cx),
}
}
}
Loading
Loading