From 57ef2f68fbb55bad95f257016386a641ca55a20b Mon Sep 17 00:00:00 2001 From: Benjamin Saunders Date: Sat, 28 Apr 2018 19:06:33 -0700 Subject: [PATCH] tokio: ergonomics and documentation --- quicr/examples/client.rs | 5 ++--- quicr/examples/server.rs | 4 +--- quicr/src/lib.rs | 39 ++++++++++++++++++++++++++++++++++----- 3 files changed, 37 insertions(+), 11 deletions(-) diff --git a/quicr/examples/client.rs b/quicr/examples/client.rs index 4641c7501..c2d7bd50a 100644 --- a/quicr/examples/client.rs +++ b/quicr/examples/client.rs @@ -9,7 +9,7 @@ extern crate slog_term; extern crate futures; extern crate url; -use std::net::{UdpSocket, ToSocketAddrs}; +use std::net::ToSocketAddrs; use std::io::{self, Write}; use futures::Future; @@ -37,7 +37,6 @@ fn run(log: Logger) -> Result<()> { let url = Url::parse(&::std::env::args().nth(1).ok_or(format_err!("missing address argument"))?)?; let remote = url.with_default_port(|_| Ok(4433))?.to_socket_addrs()?.next().ok_or(format_err!("couldn't resolve to an address"))?; - let socket = UdpSocket::bind("[::]:0")?; let mut protocols = Vec::new(); const PROTO: &[u8] = b"hq-11"; protocols.push(PROTO.len() as u8); @@ -55,7 +54,7 @@ fn run(log: Logger) -> Result<()> { protocols, ..quicr::Config::default() }) - .from_std(socket)?; + .bind("[::]:0")?; let mut executor = CurrentThread::new_with_park(timer); let request = format!("GET {}\r\n", url.path()); diff --git a/quicr/examples/server.rs b/quicr/examples/server.rs index 25ebc5bb3..5e1b4609c 100644 --- a/quicr/examples/server.rs +++ b/quicr/examples/server.rs @@ -10,7 +10,6 @@ extern crate futures; extern crate rand; extern crate openssl; -use std::net::UdpSocket; use std::fs::File; use std::io::Read; use std::fmt; @@ -69,7 +68,6 @@ fn run(log: Logger) -> Result<()> { let root = Rc::new(Path::new(&root).to_owned()); if !root.exists() { bail!("root path does not exist"); } - let socket = UdpSocket::bind("[::]:4433")?; let mut protocols = Vec::new(); const PROTO: &[u8] = b"hq-11"; protocols.push(PROTO.len() as u8); @@ -103,7 +101,7 @@ fn run(log: Logger) -> Result<()> { ..quicr::Config::default() }) .listen(quicr::ListenConfig { private_key: &key, cert: &cert, state: rand::random() }) - .from_std(socket)?; + .bind("[::]:4433")?; let mut executor = CurrentThread::new_with_park(timer); executor.spawn(incoming.for_each(move |conn| { diff --git a/quicr/src/lib.rs b/quicr/src/lib.rs index f6eac23ca..5429cf32e 100644 --- a/quicr/src/lib.rs +++ b/quicr/src/lib.rs @@ -4,7 +4,35 @@ //! head-of-line blocking, poor security, slow handshakes, and inefficient congestion control. This crate provides a //! portable userspace implementation. //! -//! The entry point of this crate is the `Endpoint`. +//! The entry point of this crate is the [`Endpoint`](struct.Endpoint.html). +//! +//! The futures and streams defined in this crate are not `Send` because they necessarily share state with eachother. As +//! a result, they must be spawned using an executor that operates on the same thread that they were constructed, such +//! as with `tokio::executor::current_thread`. The standard tokio runtime offloads futures to a threadpool, so its +//! components must instead be pieced together by hand as follows: +//! +//! ``` +//! extern crate tokio; +//! extern crate tokio_timer; +//! extern crate quicr; +//! extern crate futures; +//! +//! use futures::Future; +//! +//! fn main() { +//! let reactor = tokio::reactor::Reactor::new().unwrap(); +//! let reactor_handle = reactor.handle(); +//! let timer = tokio_timer::Timer::new(reactor); +//! let timer_handle = timer.handle(); +//! let mut executor = tokio::executor::current_thread::CurrentThread::new_with_park(timer); +//! +//! let (endpoint, driver, _) = quicr::Endpoint::new() +//! .reactor(&reactor_handle).timer(timer_handle) +//! .bind("[::]:0").unwrap(); +//! executor.spawn(driver.map_err(|e| panic!("IO error: {}", e))); +//! // ... +//! } +//! ``` #![warn(missing_docs)] @@ -23,7 +51,7 @@ extern crate failure; extern crate bytes; use std::{io, mem}; -use std::net::{SocketAddr, SocketAddrV6}; +use std::net::{SocketAddr, SocketAddrV6, ToSocketAddrs}; use std::rc::Rc; use std::cell::RefCell; use std::collections::VecDeque; @@ -161,7 +189,7 @@ impl<'a> EndpointBuilder<'a> { pub fn listen(mut self, config: ListenConfig<'a>) -> Self { self.listen = Some(config); self } pub fn config(mut self, config: Config) -> Self { self.config = config; self } - pub fn from_std(self, socket: std::net::UdpSocket) -> Result<(Endpoint, Driver, Incoming), Error> { + pub fn from_socket(self, socket: std::net::UdpSocket) -> Result<(Endpoint, Driver, Incoming), Error> { let reactor = if let Some(x) = self.reactor { Cow::Borrowed(x) } else { Cow::Owned(tokio_reactor::Handle::current()) }; let socket = UdpSocket::from_std(socket, &reactor)?; let (send, recv) = mpsc::unbounded(); @@ -179,9 +207,10 @@ impl<'a> EndpointBuilder<'a> { })); Ok((Endpoint(rc.clone()), Driver(rc), recv)) } - pub fn bind(self, addr: &SocketAddr) -> Result<(Endpoint, Driver, Incoming), Error> { + + pub fn bind(self, addr: T) -> Result<(Endpoint, Driver, Incoming), Error> { let socket = std::net::UdpSocket::bind(addr)?; - self.from_std(socket) + self.from_socket(socket) } }