Skip to content

Commit

Permalink
ex: Use smol-hyper in hyper-server example
Browse files Browse the repository at this point in the history 
Signed-off-by: John Nunley <dev@notgull.net>
  • Loading branch information
@notgull
notgull committed Jan 16, 2024
1 parent f3054c3 commit 490c4c6
Show file tree
Hide file tree
Showing 2 changed files with 86 additions and 140 deletions.
7 changes: 3 additions & 4 deletions examples/hyper-client.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
use std::convert::TryInto;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use anyhow::{bail, Context as _, Result};
Expand All @@ -23,7 +22,7 @@ use smol_macros::main;

/// Sends a request and fetches the response.
async fn fetch(
ex: &Arc<Executor<'static>>,
ex: &Executor<'static>,
req: Request<Empty<&'static [u8]>>,
) -> Result<Response<Incoming>> {
// Connect to the HTTP server.
Expand Down Expand Up @@ -66,7 +65,7 @@ async fn fetch(
}

#[apply(main!)]
async fn main(ex: Arc<Executor<'static>>) -> Result<()> {
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()
Expand All @@ -78,7 +77,7 @@ async fn main(ex: Arc<Executor<'static>>) -> Result<()> {
.body(Empty::new())?;

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

// Read the message body.
Expand Down
219 changes: 83 additions & 136 deletions examples/hyper-server.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,111 +13,97 @@
//!
//! Refer to `README.md` to see how to the TLS certificate was generated.
use std::net::{Shutdown, TcpListener, TcpStream};
use std::net::{TcpListener, TcpStream};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use anyhow::{Error, Result};
use anyhow::Result;
use async_native_tls::{Identity, TlsAcceptor, TlsStream};
use hyper::service::{make_service_fn, service_fn};
use hyper::{Body, Request, Response, Server};
use smol::{future, io, prelude::*, Async};
use http_body_util::Full;
use hyper::body::Incoming;
use hyper::service::service_fn;
use hyper::{Request, Response};
use macro_rules_attribute::apply;
use smol::{future, io, prelude::*, Async, Executor};
use smol_hyper::rt::{FuturesIo, SmolTimer};
use smol_macros::main;

/// Serves a request and returns a response.
async fn serve(req: Request<Body>, host: String) -> Result<Response<Body>> {
println!("Serving {}{}", host, req.uri());
Ok(Response::new(Body::from("Hello from hyper!")))
async fn serve(req: Request<Incoming>) -> Result<Response<Full<&'static [u8]>>> {
println!("Serving {}", req.uri());
Ok(Response::new(Full::new("Hello from hyper!".as_bytes())))
}

/// Handle a new client.
async fn handle_client(client: Async<TcpStream>, tls: Option<TlsAcceptor>) -> Result<()> {
// Wrap it in TLS if necessary.
let client = match &tls {
None => SmolStream::Plain(client),
Some(tls) => {
// In case of HTTPS, establish a secure TLS connection.
SmolStream::Tls(tls.accept(client).await?)
}
};

// Build the server.
hyper::server::conn::http1::Builder::new()
.timer(SmolTimer::new())
.serve_connection(FuturesIo::new(client), service_fn(serve))
.await?;

Ok(())
}

/// Listens for incoming connections and serves them.
async fn listen(listener: Async<TcpListener>, tls: Option<TlsAcceptor>) -> Result<()> {
async fn listen(
ex: &Arc<Executor<'static>>,
listener: Async<TcpListener>,
tls: Option<TlsAcceptor>,
) -> Result<()> {
// Format the full host address.
let host = &match tls {
None => format!("http://{}", listener.get_ref().local_addr()?),
Some(_) => format!("https://{}", listener.get_ref().local_addr()?),
};
println!("Listening on {}", host);

// Start a hyper server.
Server::builder(SmolListener::new(&listener, tls))
.executor(SmolExecutor)
.serve(make_service_fn(move |_| {
let host = host.clone();
async { Ok::<_, Error>(service_fn(move |req| serve(req, host.clone()))) }
}))
.await?;
loop {
// Wait for a new client.
let (client, _) = listener.accept().await?;

Ok(())
// Spawn a task to handle this connection.
ex.spawn({
let tls = tls.clone();
async move {
if let Err(e) = handle_client(client, tls).await {
println!("Error while handling client: {}", e);
}
}
})
.detach();
}
}

fn main() -> Result<()> {
#[apply(main!)]
async fn main(ex: &Arc<Executor<'static>>) -> Result<()> {
// Initialize TLS with the local certificate, private key, and password.
let identity = Identity::from_pkcs12(include_bytes!("identity.pfx"), "password")?;
let tls = TlsAcceptor::from(native_tls::TlsAcceptor::new(identity)?);

// Start HTTP and HTTPS servers.
smol::block_on(async {
let http = listen(Async::<TcpListener>::bind(([127, 0, 0, 1], 8000))?, None);
let https = listen(
Async::<TcpListener>::bind(([127, 0, 0, 1], 8001))?,
Some(tls),
);
future::try_zip(http, https).await?;
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();
}
}

/// Listens for incoming connections.
struct SmolListener<'a> {
tls: Option<TlsAcceptor>,
incoming: Pin<Box<dyn Stream<Item = io::Result<Async<TcpStream>>> + Send + 'a>>,
}

impl<'a> SmolListener<'a> {
fn new(listener: &'a Async<TcpListener>, tls: Option<TlsAcceptor>) -> Self {
Self {
incoming: Box::pin(listener.incoming()),
tls,
}
}
}

impl hyper::server::accept::Accept for SmolListener<'_> {
type Conn = SmolStream;
type Error = Error;

fn poll_accept(
mut self: Pin<&mut Self>,
cx: &mut Context,
) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
let stream = smol::ready!(self.incoming.as_mut().poll_next(cx)).unwrap()?;

let stream = match &self.tls {
None => SmolStream::Plain(stream),
Some(tls) => {
// In case of HTTPS, start establishing a secure TLS connection.
let tls = tls.clone();
SmolStream::Handshake(Box::pin(async move {
tls.accept(stream).await.map_err(|err| {
println!("Failed to establish secure TLS connection: {:#?}", err);
io::Error::new(io::ErrorKind::Other, Box::new(err))
})
}))
}
};

Poll::Ready(Some(Ok(stream)))
}
let http = listen(
ex,
Async::<TcpListener>::bind(([127, 0, 0, 1], 8000))?,
None,
);
let https = listen(
ex,
Async::<TcpListener>::bind(([127, 0, 0, 1], 8001))?,
Some(tls),
);
future::try_zip(http, https).await?;
Ok(())
}

/// A TCP or TCP+TLS connection.
Expand All @@ -127,83 +113,44 @@ enum SmolStream {

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

/// A TCP connection that is in process of getting secured by TLS.
#[allow(clippy::type_complexity)]
Handshake(Pin<Box<dyn Future<Output = io::Result<TlsStream<Async<TcpStream>>>> + Send>>),
}

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 {
impl AsyncRead for SmolStream {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<io::Result<()>> {
loop {
match &mut *self {
SmolStream::Plain(s) => {
return Pin::new(s)
.poll_read(cx, buf.initialize_unfilled())
.map_ok(|size| {
buf.advance(size);
});
}
SmolStream::Tls(s) => {
return Pin::new(s)
.poll_read(cx, buf.initialize_unfilled())
.map_ok(|size| {
buf.advance(size);
});
}
SmolStream::Handshake(f) => {
let s = smol::ready!(f.as_mut().poll(cx))?;
*self = SmolStream::Tls(s);
}
}
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
match &mut *self {
Self::Plain(s) => Pin::new(s).poll_read(cx, buf),
Self::Tls(s) => Pin::new(s).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>> {
loop {
match &mut *self {
SmolStream::Plain(s) => return Pin::new(s).poll_write(cx, buf),
SmolStream::Tls(s) => return Pin::new(s).poll_write(cx, buf),
SmolStream::Handshake(f) => {
let s = smol::ready!(f.as_mut().poll(cx))?;
*self = SmolStream::Tls(s);
}
}
match &mut *self {
Self::Plain(s) => Pin::new(s).poll_write(cx, buf),
Self::Tls(s) => Pin::new(s).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::Handshake(_) => Poll::Ready(Ok(())),
Self::Plain(s) => Pin::new(s).poll_close(cx),
Self::Tls(s) => Pin::new(s).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::Handshake(_) => Poll::Ready(Ok(())),
Self::Plain(s) => Pin::new(s).poll_close(cx),
Self::Tls(s) => Pin::new(s).poll_close(cx),
}
}
}

0 comments on commit 490c4c6

Please sign in to comment.