forked/h2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update to Tokio 0.2 (#428)

Browse Source
This commit is contained in:
Sean McArthur
2019-11-27 14:53:57 -08:00
committed by GitHub
parent 37b66e8981
commit 4398e169e8
53 changed files with 473 additions and 972 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.github/workflows/CI.yml vendored
View File

@@ -47,9 +47,9 @@ jobs:
# TODO: Change it to stable after Rust 1.38 release
run: ./ci/h2spec.sh
if: matrix.rust == 'nightly'
- name: Check minimal versions
run: cargo clean; cargo update -Zminimal-versions; cargo check
if: matrix.rust == 'nightly'
#- name: Check minimal versions
# run: cargo clean; cargo update -Zminimal-versions; cargo check
# if: matrix.rust == 'nightly'
publish_docs:
name: Publish Documentation

30
Cargo.toml
View File

@@ -38,18 +38,16 @@ members = [
]
[dependencies]
futures-core-preview = "=0.3.0-alpha.19"
futures-sink-preview = "=0.3.0-alpha.19"
futures-util-preview = "=0.3.0-alpha.19"
tokio-codec = "=0.2.0-alpha.6"
tokio-io = { version = "=0.2.0-alpha.6", features = ["util"] }
tokio-sync = "=0.2.0-alpha.6"
bytes = "0.4.7"
http = "0.1.8"
futures-core = "0.3"
futures-sink = "0.3"
futures-util = { version = "0.3", default-features = false, features = [] }
tokio-util = { version = "0.2", features = ["codec"] }
tokio = { version = "0.2", features = ["io-util", "sync"] }
bytes = "0.5.2"
http = { git = "https://github.com/hyperium/http" } #"0.1.8"
log = "0.4.1"
fnv = "1.0.5"
slab = "0.4.0"
string = "0.2"
indexmap = "1.0"
[dev-dependencies]
@@ -64,10 +62,12 @@ walkdir = "1.0.0"
serde = "1.0.0"
serde_json = "1.0.0"
# Akamai example
tokio = "=0.2.0-alpha.6"
# Examples
tokio = { version = "0.2", features = ["dns", "macros", "rt-core", "tcp"] }
env_logger = { version = "0.5.3", default-features = false }
rustls = "0.16"
tokio-rustls = "0.12.0-alpha.4"
webpki = "0.21"
webpki-roots = "0.17"
# TODO: re-enable when tokio-rustls is updated
#rustls = "0.16"
#tokio-rustls = "0.12.0-alpha.4"
#webpki = "0.21"
#webpki-roots = "0.17"

3
examples/akamai.rs
View File

@@ -1,3 +1,5 @@
fn main() {}
/* TODO: re-enable when tokio-rustls is updated
use h2::client;
use http::{Method, Request};
use tokio::net::TcpStream;
@@ -73,3 +75,4 @@ pub async fn main() -> Result<(), Box<dyn Error>> {
}
Ok(())
}
*/

14
examples/server.rs
View File

@@ -1,13 +1,11 @@
use h2::server;
use bytes::*;
use http::{Response, StatusCode};
use std::error::Error;
use bytes::Bytes;
use h2::server;
use tokio::net::{TcpListener, TcpStream};
#[tokio::main]
pub async fn main() -> Result<(), Box<dyn Error>> {
async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
let _ = env_logger::try_init();
let mut listener = TcpListener::bind("127.0.0.1:5928").await?;
@@ -25,14 +23,14 @@ pub async fn main() -> Result<(), Box<dyn Error>> {
}
}
async fn handle(socket: TcpStream) -> Result<(), Box<dyn Error>> {
async fn handle(socket: TcpStream) -> Result<(), Box<dyn Error + Send + Sync>> {
let mut connection = server::handshake(socket).await?;
println!("H2 connection bound");
while let Some(result) = connection.accept().await {
let (request, mut respond) = result?;
println!("GOT request: {:?}", request);
let response = Response::builder().status(StatusCode::OK).body(()).unwrap();
let response = http::Response::new(());
let mut send = respond.send_response(response, false)?;

70
src/client.rs
View File

@@ -140,7 +140,7 @@ use crate::frame::{Headers, Pseudo, Reason, Settings, StreamId};
use crate::proto;
use crate::{FlowControl, PingPong, RecvStream, SendStream};
use bytes::{Bytes, IntoBuf};
use bytes::{Buf, Bytes};
use http::{uri, HeaderMap, Method, Request, Response, Version};
use std::fmt;
use std::future::Future;
@@ -148,7 +148,7 @@ use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use std::usize;
use tokio_io::{AsyncRead, AsyncWrite, AsyncWriteExt};
use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
/// Initializes new HTTP/2.0 streams on a connection by sending a request.
///
@@ -171,15 +171,15 @@ use tokio_io::{AsyncRead, AsyncWrite, AsyncWriteExt};
/// [`Connection`]: struct.Connection.html
/// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html
/// [`Error`]: ../struct.Error.html
pub struct SendRequest<B: IntoBuf> {
inner: proto::Streams<B::Buf, Peer>,
pub struct SendRequest<B: Buf> {
inner: proto::Streams<B, Peer>,
pending: Option<proto::OpaqueStreamRef>,
}
/// Returns a `SendRequest` instance once it is ready to send at least one
/// request.
#[derive(Debug)]
pub struct ReadySendRequest<B: IntoBuf> {
pub struct ReadySendRequest<B: Buf> {
inner: Option<SendRequest<B>>,
}
@@ -208,7 +208,7 @@ pub struct ReadySendRequest<B: IntoBuf> {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client;
/// # use h2::client::*;
/// #
@@ -227,7 +227,7 @@ pub struct ReadySendRequest<B: IntoBuf> {
/// # pub fn main() {}
/// ```
#[must_use = "futures do nothing unless polled"]
pub struct Connection<T, B: IntoBuf = Bytes> {
pub struct Connection<T, B: Buf = Bytes> {
inner: proto::Connection<T, Peer, B>,
}
@@ -286,7 +286,7 @@ pub struct PushPromises {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use bytes::Bytes;
/// #
@@ -336,8 +336,7 @@ pub(crate) struct Peer;
impl<B> SendRequest<B>
where
B: IntoBuf + Unpin,
B::Buf: Unpin + 'static,
B: Buf + Unpin + 'static,
{
/// Returns `Ready` when the connection can initialize a new HTTP/2.0
/// stream.
@@ -521,7 +520,7 @@ where
impl<B> fmt::Debug for SendRequest<B>
where
B: IntoBuf,
B: Buf,
{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_struct("SendRequest").finish()
@@ -530,7 +529,7 @@ where
impl<B> Clone for SendRequest<B>
where
B: IntoBuf,
B: Buf,
{
fn clone(&self) -> Self {
SendRequest {
@@ -543,7 +542,7 @@ where
#[cfg(feature = "unstable")]
impl<B> SendRequest<B>
where
B: IntoBuf,
B: Buf,
{
/// Returns the number of active streams.
///
@@ -567,8 +566,7 @@ where
impl<B> Future for ReadySendRequest<B>
where
B: IntoBuf + Unpin,
B::Buf: Unpin + 'static,
B: Buf + Unpin + 'static,
{
type Output = Result<SendRequest<B>, crate::Error>;
@@ -595,7 +593,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use bytes::Bytes;
/// #
@@ -637,7 +635,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use bytes::Bytes;
/// #
@@ -672,7 +670,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use bytes::Bytes;
/// #
@@ -706,7 +704,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use bytes::Bytes;
/// #
@@ -746,7 +744,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use bytes::Bytes;
/// #
@@ -795,7 +793,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use bytes::Bytes;
/// #
@@ -836,7 +834,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use bytes::Bytes;
/// #
@@ -881,7 +879,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use bytes::Bytes;
/// #
@@ -926,7 +924,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use std::time::Duration;
/// # use bytes::Bytes;
@@ -964,7 +962,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use std::time::Duration;
/// # use bytes::Bytes;
@@ -1023,7 +1021,7 @@ impl Builder {
/// Basic usage:
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// # use bytes::Bytes;
/// #
@@ -1044,7 +1042,7 @@ impl Builder {
/// type will be `&'static [u8]`.
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client::*;
/// #
/// # async fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
@@ -1065,8 +1063,7 @@ impl Builder {
) -> impl Future<Output = Result<(SendRequest<B>, Connection<T, B>), crate::Error>>
where
T: AsyncRead + AsyncWrite + Unpin,
B: IntoBuf + Unpin,
B::Buf: Unpin + 'static,
B: Buf + Unpin + 'static,
{
Connection::handshake2(io, self.clone())
}
@@ -1098,7 +1095,7 @@ impl Default for Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::client;
/// # use h2::client::*;
/// #
@@ -1126,8 +1123,7 @@ where
impl<T, B> Connection<T, B>
where
T: AsyncRead + AsyncWrite + Unpin,
B: IntoBuf + Unpin,
B::Buf: Unpin,
B: Buf + Unpin + 'static,
{
async fn handshake2(
mut io: T,
@@ -1233,8 +1229,7 @@ where
impl<T, B> Future for Connection<T, B>
where
T: AsyncRead + AsyncWrite + Unpin,
B: IntoBuf + Unpin,
B::Buf: Unpin,
B: Buf + Unpin + 'static,
{
type Output = Result<(), crate::Error>;
@@ -1248,8 +1243,7 @@ impl<T, B> fmt::Debug for Connection<T, B>
where
T: AsyncRead + AsyncWrite,
T: fmt::Debug,
B: fmt::Debug + IntoBuf,
B::Buf: fmt::Debug,
B: fmt::Debug + Buf,
{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt::Debug::fmt(&self.inner, fmt)
@@ -1459,10 +1453,10 @@ impl proto::Peer for Peer {
) -> Result<Self::Poll, RecvError> {
let mut b = Response::builder();
b.version(Version::HTTP_2);
b = b.version(Version::HTTP_2);
if let Some(status) = pseudo.status {
b.status(status);
b = b.status(status);
}
let mut response = match b.body(()) {

6
src/codec/framed_read.rs
View File

@@ -14,9 +14,9 @@ use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio_codec::FramedRead as InnerFramedRead;
use tokio_codec::{LengthDelimitedCodec, LengthDelimitedCodecError};
use tokio_io::AsyncRead;
use tokio::io::AsyncRead;
use tokio_util::codec::FramedRead as InnerFramedRead;
use tokio_util::codec::{LengthDelimitedCodec, LengthDelimitedCodecError};
// 16 MB "sane default" taken from golang http2
const DEFAULT_SETTINGS_MAX_HEADER_LIST_SIZE: usize = 16 << 20;

28
src/codec/framed_write.rs
View File

@@ -3,13 +3,24 @@ use crate::codec::UserError::*;
use crate::frame::{self, Frame, FrameSize};
use crate::hpack;
use bytes::{Buf, BufMut, BytesMut};
use bytes::{
buf::{BufExt, BufMutExt},
Buf, BufMut, BytesMut,
};
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio_io::{AsyncRead, AsyncWrite};
use tokio::io::{AsyncRead, AsyncWrite};
use std::io::{self, Cursor};
// A macro to get around a method needing to borrow &mut self
macro_rules! limited_write_buf {
($self:expr) => {{
let limit = $self.max_frame_size() + frame::HEADER_LEN;
$self.buf.get_mut().limit(limit)
}};
}
#[derive(Debug)]
pub struct FramedWrite<T, B> {
/// Upstream `AsyncWrite`
@@ -126,12 +137,14 @@ where
}
}
Frame::Headers(v) => {
if let Some(continuation) = v.encode(&mut self.hpack, self.buf.get_mut()) {
let mut buf = limited_write_buf!(self);
if let Some(continuation) = v.encode(&mut self.hpack, &mut buf) {
self.next = Some(Next::Continuation(continuation));
}
}
Frame::PushPromise(v) => {
if let Some(continuation) = v.encode(&mut self.hpack, self.buf.get_mut()) {
let mut buf = limited_write_buf!(self);
if let Some(continuation) = v.encode(&mut self.hpack, &mut buf) {
self.next = Some(Next::Continuation(continuation));
}
}
@@ -177,7 +190,7 @@ where
match self.next {
Some(Next::Data(ref mut frame)) => {
log::trace!(" -> queued data frame");
let mut buf = Buf::by_ref(&mut self.buf).chain(frame.payload_mut());
let mut buf = (&mut self.buf).chain(frame.payload_mut());
ready!(Pin::new(&mut self.inner).poll_write_buf(cx, &mut buf))?;
}
_ => {
@@ -200,7 +213,8 @@ where
}
Some(Next::Continuation(frame)) => {
// Buffer the continuation frame, then try to write again
if let Some(continuation) = frame.encode(&mut self.hpack, self.buf.get_mut()) {
let mut buf = limited_write_buf!(self);
if let Some(continuation) = frame.encode(&mut self.hpack, &mut buf) {
// We previously had a CONTINUATION, and after encoding
// it, we got *another* one? Let's just double check
// that at least some progress is being made...
@@ -268,7 +282,7 @@ impl<T, B> FramedWrite<T, B> {
}
impl<T: AsyncRead + Unpin, B: Unpin> AsyncRead for FramedWrite<T, B> {
unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [std::mem::MaybeUninit<u8>]) -> bool {
self.inner.prepare_uninitialized_buffer(buf)
}

6
src/codec/mod.rs
View File

@@ -14,8 +14,8 @@ use futures_core::Stream;
use futures_sink::Sink;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio_codec::length_delimited;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio::io::{AsyncRead, AsyncWrite};
use tokio_util::codec::length_delimited;
use std::io;
@@ -186,7 +186,7 @@ where
}
// TODO: remove (or improve) this
impl<T> From<T> for Codec<T, ::std::io::Cursor<::bytes::Bytes>>
impl<T> From<T> for Codec<T, bytes::Bytes>
where
T: AsyncRead + AsyncWrite + Unpin,
{

6
src/frame/go_away.rs
View File

@@ -41,7 +41,7 @@ impl GoAway {
let (last_stream_id, _) = StreamId::parse(&payload[..4]);
let error_code = unpack_octets_4!(payload, 4, u32);
let debug_data = Bytes::from(&payload[8..]);
let debug_data = Bytes::copy_from_slice(&payload[8..]);
Ok(GoAway {
last_stream_id,
@@ -54,8 +54,8 @@ impl GoAway {
log::trace!("encoding GO_AWAY; code={:?}", self.error_code);
let head = Head::new(Kind::GoAway, 0, StreamId::zero());
head.encode(8, dst);
dst.put_u32_be(self.last_stream_id.into());
dst.put_u32_be(self.error_code.into());
dst.put_u32(self.last_stream_id.into());
dst.put_u32(self.error_code.into());
}
}

4
src/frame/head.rs
View File

@@ -66,10 +66,10 @@ impl Head {
pub fn encode<T: BufMut>(&self, payload_len: usize, dst: &mut T) {
debug_assert!(self.encode_len() <= dst.remaining_mut());
dst.put_uint_be(payload_len as u64, 3);
dst.put_uint(payload_len as u64, 3);
dst.put_u8(self.kind as u8);
dst.put_u8(self.flag);
dst.put_u32_be(self.stream_id.into());
dst.put_u32(self.stream_id.into());
}
}

71
src/frame/headers.rs
View File

@@ -1,16 +1,17 @@
use super::{util, StreamDependency, StreamId};
use crate::frame::{Error, Frame, Head, Kind};
use crate::hpack;
use crate::hpack::{self, BytesStr};
use http::header::{self, HeaderName, HeaderValue};
use http::{uri, HeaderMap, Method, Request, StatusCode, Uri};
use bytes::{Bytes, BytesMut};
use string::String;
use std::fmt;
use std::io::Cursor;
type EncodeBuf<'a> = bytes::buf::ext::Limit<&'a mut BytesMut>;
// Minimum MAX_FRAME_SIZE is 16kb, so save some arbitrary space for frame
// head and other header bits.
const MAX_HEADER_LENGTH: usize = 1024 * 16 - 100;
@@ -67,9 +68,9 @@ pub struct Continuation {
pub struct Pseudo {
// Request
pub method: Option<Method>,
pub scheme: Option<String<Bytes>>,
pub authority: Option<String<Bytes>>,
pub path: Option<String<Bytes>>,
pub scheme: Option<BytesStr>,
pub authority: Option<BytesStr>,
pub path: Option<BytesStr>,
// Response
pub status: Option<StatusCode>,
@@ -261,7 +262,11 @@ impl Headers {
self.header_block.fields
}
pub fn encode(self, encoder: &mut hpack::Encoder, dst: &mut BytesMut) -> Option<Continuation> {
pub fn encode(
self,
encoder: &mut hpack::Encoder,
dst: &mut EncodeBuf<'_>,
) -> Option<Continuation> {
// At this point, the `is_end_headers` flag should always be set
debug_assert!(self.flags.is_end_headers());
@@ -465,7 +470,11 @@ impl PushPromise {
self.header_block.is_over_size
}
pub fn encode(self, encoder: &mut hpack::Encoder, dst: &mut BytesMut) -> Option<Continuation> {
pub fn encode(
self,
encoder: &mut hpack::Encoder,
dst: &mut EncodeBuf<'_>,
) -> Option<Continuation> {
use bytes::BufMut;
// At this point, the `is_end_headers` flag should always be set
@@ -477,7 +486,7 @@ impl PushPromise {
self.header_block
.into_encoding()
.encode(&head, encoder, dst, |dst| {
dst.put_u32_be(promised_id.into());
dst.put_u32(promised_id.into());
})
}
@@ -515,7 +524,11 @@ impl Continuation {
Head::new(Kind::Continuation, END_HEADERS, self.stream_id)
}
pub fn encode(self, encoder: &mut hpack::Encoder, dst: &mut BytesMut) -> Option<Continuation> {
pub fn encode(
self,
encoder: &mut hpack::Encoder,
dst: &mut EncodeBuf<'_>,
) -> Option<Continuation> {
// Get the CONTINUATION frame head
let head = self.head();
@@ -542,7 +555,7 @@ impl Pseudo {
method: Some(method),
scheme: None,
authority: None,
path: Some(to_string(path)),
path: Some(unsafe { BytesStr::from_utf8_unchecked(path) }),
status: None,
};
@@ -556,7 +569,7 @@ impl Pseudo {
// If the URI includes an authority component, add it to the pseudo
// headers
if let Some(authority) = parts.authority {
pseudo.set_authority(to_string(authority.into()));
pseudo.set_authority(unsafe { BytesStr::from_utf8_unchecked(authority.into()) });
}
pseudo
@@ -573,18 +586,14 @@ impl Pseudo {
}
pub fn set_scheme(&mut self, scheme: uri::Scheme) {
self.scheme = Some(to_string(scheme.into()));
self.scheme = Some(unsafe { BytesStr::from_utf8_unchecked(scheme.into()) });
}
pub fn set_authority(&mut self, authority: String<Bytes>) {
pub fn set_authority(&mut self, authority: BytesStr) {
self.authority = Some(authority);
}
}
fn to_string(src: Bytes) -> String<Bytes> {
unsafe { String::from_utf8_unchecked(src) }
}
// ===== impl EncodingHeaderBlock =====
impl EncodingHeaderBlock {
@@ -592,20 +601,20 @@ impl EncodingHeaderBlock {
mut self,
head: &Head,
encoder: &mut hpack::Encoder,
dst: &mut BytesMut,
dst: &mut EncodeBuf<'_>,
f: F,
) -> Option<Continuation>
where
F: FnOnce(&mut BytesMut),
F: FnOnce(&mut EncodeBuf<'_>),
{
let head_pos = dst.len();
let head_pos = dst.get_ref().len();
// At this point, we don't know how big the h2 frame will be.
// So, we write the head with length 0, then write the body, and
// finally write the length once we know the size.
head.encode(0, dst);
let payload_pos = dst.len();
let payload_pos = dst.get_ref().len();
f(dst);
@@ -622,19 +631,19 @@ impl EncodingHeaderBlock {
};
// Compute the header block length
let payload_len = (dst.len() - payload_pos) as u64;
let payload_len = (dst.get_ref().len() - payload_pos) as u64;
// Write the frame length
let payload_len_be = payload_len.to_be_bytes();
assert!(payload_len_be[0..5].iter().all(|b| *b == 0));
(&mut dst[head_pos..head_pos + 3]).copy_from_slice(&payload_len_be[5..]);
(dst.get_mut()[head_pos..head_pos + 3]).copy_from_slice(&payload_len_be[5..]);
if continuation.is_some() {
// There will be continuation frames, so the `is_end_headers` flag
// must be unset
debug_assert!(dst[head_pos + 4] & END_HEADERS == END_HEADERS);
debug_assert!(dst.get_ref()[head_pos + 4] & END_HEADERS == END_HEADERS);
dst[head_pos + 4] -= END_HEADERS;
dst.get_mut()[head_pos + 4] -= END_HEADERS;
}
continuation
@@ -962,15 +971,3 @@ impl HeaderBlock {
fn decoded_header_size(name: usize, value: usize) -> usize {
name + value + 32
}
// Stupid hack to make the set_pseudo! macro happy, since all other values
// have a method `as_str` except for `String<Bytes>`.
trait AsStr {
fn as_str(&self) -> &str;
}
impl AsStr for String<Bytes> {
fn as_str(&self) -> &str {
self
}
}

3
src/frame/mod.rs
View File

@@ -53,6 +53,9 @@ pub use self::settings::Settings;
pub use self::stream_id::{StreamId, StreamIdOverflow};
pub use self::window_update::WindowUpdate;
#[cfg(feature = "unstable")]
pub use crate::hpack::BytesStr;
// Re-export some constants
pub use self::settings::{

4
src/frame/ping.rs
View File

@@ -1,5 +1,5 @@
use crate::frame::{Error, Frame, Head, Kind, StreamId};
use bytes::{Buf, BufMut, IntoBuf};
use bytes::BufMut;
const ACK_FLAG: u8 = 0x1;
@@ -71,7 +71,7 @@ impl Ping {
}
let mut payload = [0; 8];
bytes.into_buf().copy_to_slice(&mut payload);
payload.copy_from_slice(bytes);
// The PING frame defines the following flags:
//

2
src/frame/reset.rs
View File

@@ -45,7 +45,7 @@ impl Reset {
);
let head = Head::new(Kind::Reset, 0, self.stream_id);
head.encode(4, dst);
dst.put_u32_be(self.error_code.into());
dst.put_u32(self.error_code.into());
}
}

4
src/frame/settings.rs
View File

@@ -314,8 +314,8 @@ impl Setting {
MaxHeaderListSize(v) => (6, v),
};
dst.put_u16_be(kind);
dst.put_u32_be(val);
dst.put_u16(kind);
dst.put_u32(val);
}
}

2
src/frame/window_update.rs
View File

@@ -51,7 +51,7 @@ impl WindowUpdate {
log::trace!("encoding WINDOW_UPDATE; id={:?}", self.stream_id);
let head = Head::new(Kind::WindowUpdate, 0, self.stream_id);
head.encode(4, dst);
dst.put_u32_be(self.size_increment);
dst.put_u32(self.size_increment);
}
}

14
src/hpack/decoder.rs
View File

@@ -1,11 +1,10 @@
use super::{huffman, Header};
use super::{header::BytesStr, huffman, Header};
use crate::frame;
use bytes::{Buf, Bytes, BytesMut};
use http::header;
use http::method::{self, Method};
use http::status::{self, StatusCode};
use string::String;
use std::cmp;
use std::collections::VecDeque;
@@ -314,7 +313,7 @@ impl Decoder {
if huff {
let ret = {
let raw = &buf.bytes()[..len];
huffman::decode(raw, &mut self.buffer).map(Into::into)
huffman::decode(raw, &mut self.buffer).map(BytesMut::freeze)
};
buf.advance(len);
@@ -785,8 +784,8 @@ pub fn get_static(idx: usize) -> Header {
}
}
fn from_static(s: &'static str) -> String<Bytes> {
unsafe { String::from_utf8_unchecked(Bytes::from_static(s.as_bytes())) }
fn from_static(s: &'static str) -> BytesStr {
unsafe { BytesStr::from_utf8_unchecked(Bytes::from_static(s.as_bytes())) }
}
#[cfg(test)]
@@ -823,13 +822,12 @@ mod test {
fn test_decode_indexed_larger_than_table() {
let mut de = Decoder::new(0);
let mut buf = vec![0b01000000, 0x80 | 2];
let mut buf = BytesMut::new();
buf.extend(&[0b01000000, 0x80 | 2]);
buf.extend(huff_encode(b"foo"));
buf.extend(&[0x80 | 3]);
buf.extend(huff_encode(b"bar"));
let mut buf = buf.into();
let mut res = vec![];
let _ = de
.decode(&mut Cursor::new(&mut buf), |h| {

72
src/hpack/encoder.rs
View File

@@ -1,9 +1,11 @@
use super::table::{Index, Table};
use super::{huffman, Header};
use bytes::{BufMut, BytesMut};
use bytes::{buf::ext::Limit, BufMut, BytesMut};
use http::header::{HeaderName, HeaderValue};
type DstBuf<'a> = Limit<&'a mut BytesMut>;
#[derive(Debug)]
pub struct Encoder {
table: Table,
@@ -80,16 +82,16 @@ impl Encoder {
&mut self,
resume: Option<EncodeState>,
headers: &mut I,
dst: &mut BytesMut,
dst: &mut DstBuf<'_>,
) -> Encode
where
I: Iterator<Item = Header<Option<HeaderName>>>,
{
let len = dst.len();
let pos = position(dst);
if let Err(e) = self.encode_size_updates(dst) {
if e == EncoderError::BufferOverflow {
dst.truncate(len);
rewind(dst, pos);
}
unreachable!("encode_size_updates errored");
@@ -98,7 +100,7 @@ impl Encoder {
let mut last_index = None;
if let Some(resume) = resume {
let len = dst.len();
let pos = position(dst);
let res = match resume.value {
Some(ref value) => self.encode_header_without_name(&resume.index, value, dst),
@@ -106,14 +108,14 @@ impl Encoder {
};
if res.is_err() {
dst.truncate(len);
rewind(dst, pos);
return Encode::Partial(resume);
}
last_index = Some(resume.index);
}
for header in headers {
let len = dst.len();
let pos = position(dst);
match header.reify() {
// The header has an associated name. In which case, try to
@@ -123,7 +125,7 @@ impl Encoder {
let res = self.encode_header(&index, dst);
if res.is_err() {
dst.truncate(len);
rewind(dst, pos);
return Encode::Partial(EncodeState { index, value: None });
}
@@ -143,7 +145,7 @@ impl Encoder {
);
if res.is_err() {
dst.truncate(len);
rewind(dst, pos);
return Encode::Partial(EncodeState {
index: last_index.unwrap(), // checked just above
value: Some(value),
@@ -156,7 +158,7 @@ impl Encoder {
Encode::Full
}
fn encode_size_updates(&mut self, dst: &mut BytesMut) -> Result<(), EncoderError> {
fn encode_size_updates(&mut self, dst: &mut DstBuf<'_>) -> Result<(), EncoderError> {
match self.size_update.take() {
Some(SizeUpdate::One(val)) => {
self.table.resize(val);
@@ -174,7 +176,7 @@ impl Encoder {
Ok(())
}
fn encode_header(&mut self, index: &Index, dst: &mut BytesMut) -> Result<(), EncoderError> {
fn encode_header(&mut self, index: &Index, dst: &mut DstBuf<'_>) -> Result<(), EncoderError> {
match *index {
Index::Indexed(idx, _) => {
encode_int(idx, 7, 0x80, dst)?;
@@ -225,7 +227,7 @@ impl Encoder {
&mut self,
last: &Index,
value: &HeaderValue,
dst: &mut BytesMut,
dst: &mut DstBuf<'_>,
) -> Result<(), EncoderError> {
match *last {
Index::Indexed(..)
@@ -266,7 +268,7 @@ fn encode_not_indexed(
name: usize,
value: &[u8],
sensitive: bool,
dst: &mut BytesMut,
dst: &mut DstBuf<'_>,
) -> Result<(), EncoderError> {
if sensitive {
encode_int(name, 4, 0b10000, dst)?;
@@ -282,7 +284,7 @@ fn encode_not_indexed2(
name: &[u8],
value: &[u8],
sensitive: bool,
dst: &mut BytesMut,
dst: &mut DstBuf<'_>,
) -> Result<(), EncoderError> {
if !dst.has_remaining_mut() {
return Err(EncoderError::BufferOverflow);
@@ -299,15 +301,13 @@ fn encode_not_indexed2(
Ok(())
}
fn encode_str(val: &[u8], dst: &mut BytesMut) -> Result<(), EncoderError> {
use std::io::Cursor;
fn encode_str(val: &[u8], dst: &mut DstBuf<'_>) -> Result<(), EncoderError> {
if !dst.has_remaining_mut() {
return Err(EncoderError::BufferOverflow);
}
if !val.is_empty() {
let idx = dst.len();
let idx = position(dst);
// Push a placeholder byte for the length header
dst.put_u8(0);
@@ -315,19 +315,20 @@ fn encode_str(val: &[u8], dst: &mut BytesMut) -> Result<(), EncoderError> {
// Encode with huffman
huffman::encode(val, dst)?;
let huff_len = dst.len() - (idx + 1);
let huff_len = position(dst) - (idx + 1);
if encode_int_one_byte(huff_len, 7) {
// Write the string head
dst[idx] = 0x80 | huff_len as u8;
dst.get_mut()[idx] = 0x80 | huff_len as u8;
} else {
// Write the head to a placeholer
let mut buf = [0; 8];
const PLACEHOLDER_LEN: usize = 8;
let mut buf = [0u8; PLACEHOLDER_LEN];
let head_len = {
let mut head_dst = Cursor::new(&mut buf);
let mut head_dst = &mut buf[..];
encode_int(huff_len, 7, 0x80, &mut head_dst)?;
head_dst.position() as usize
PLACEHOLDER_LEN - head_dst.remaining_mut()
};
if dst.remaining_mut() < head_len {
@@ -337,16 +338,17 @@ fn encode_str(val: &[u8], dst: &mut BytesMut) -> Result<(), EncoderError> {
// This is just done to reserve space in the destination
dst.put_slice(&buf[1..head_len]);
let written = dst.get_mut();
// Shift the header forward
for i in 0..huff_len {
let src_i = idx + 1 + (huff_len - (i + 1));
let dst_i = idx + head_len + (huff_len - (i + 1));
dst[dst_i] = dst[src_i];
written[dst_i] = written[src_i];
}
// Copy in the head
for i in 0..head_len {
dst[idx + i] = buf[i];
written[idx + i] = buf[i];
}
}
} else {
@@ -411,10 +413,19 @@ fn encode_int_one_byte(value: usize, prefix_bits: usize) -> bool {
value < (1 << prefix_bits) - 1
}
fn position(buf: &DstBuf<'_>) -> usize {
buf.get_ref().len()
}
fn rewind(buf: &mut DstBuf<'_>, pos: usize) {
buf.get_mut().truncate(pos);
}
#[cfg(test)]
mod test {
use super::*;
use crate::hpack::Header;
use bytes::buf::BufMutExt;
use http::*;
#[test]
@@ -794,7 +805,8 @@ mod test {
#[test]
fn test_nameless_header_at_resume() {
let mut encoder = Encoder::default();
let mut dst = BytesMut::from(Vec::with_capacity(15));
let max_len = 15;
let mut dst = BytesMut::with_capacity(64);
let mut input = vec![
Header::Field {
@@ -812,9 +824,9 @@ mod test {
]
.into_iter();
let resume = match encoder.encode(None, &mut input, &mut dst) {
let resume = match encoder.encode(None, &mut input, &mut (&mut dst).limit(max_len)) {
Encode::Partial(r) => r,
_ => panic!(),
_ => panic!("encode should be partial"),
};
assert_eq!(&[0x40, 0x80 | 4], &dst[0..2]);
@@ -824,7 +836,7 @@ mod test {
dst.clear();
match encoder.encode(Some(resume), &mut input, &mut dst) {
match encoder.encode(Some(resume), &mut input, &mut (&mut dst).limit(max_len)) {
Encode::Full => {}
unexpected => panic!("resume returned unexpected: {:?}", unexpected),
}
@@ -844,7 +856,7 @@ mod test {
fn encode(e: &mut Encoder, hdrs: Vec<Header<Option<HeaderName>>>) -> BytesMut {
let mut dst = BytesMut::with_capacity(1024);
e.encode(None, &mut hdrs.into_iter(), &mut dst);
e.encode(None, &mut hdrs.into_iter(), &mut (&mut dst).limit(1024));
dst
}

66
src/hpack/header.rs
View File

@@ -3,17 +3,17 @@ use super::{DecoderError, NeedMore};
use bytes::Bytes;
use http::header::{HeaderName, HeaderValue};
use http::{Method, StatusCode};
use string::{String, TryFrom};
use std::fmt;
/// HTTP/2.0 Header
#[derive(Debug, Clone, Eq, PartialEq)]
pub enum Header<T = HeaderName> {
Field { name: T, value: HeaderValue },
// TODO: Change these types to `http::uri` types.
Authority(String<Bytes>),
Authority(BytesStr),
Method(Method),
Scheme(String<Bytes>),
Path(String<Bytes>),
Scheme(BytesStr),
Path(BytesStr),
Status(StatusCode),
}
@@ -28,6 +28,10 @@ pub enum Name<'a> {
Status,
}
#[doc(hidden)]
#[derive(Clone, Eq, PartialEq, Default)]
pub struct BytesStr(Bytes);
pub fn len(name: &HeaderName, value: &HeaderValue) -> usize {
let n: &str = name.as_ref();
32 + n.len() + value.len()
@@ -60,7 +64,7 @@ impl Header {
if name[0] == b':' {
match &name[1..] {
b"authority" => {
let value = String::try_from(value)?;
let value = BytesStr::try_from(value)?;
Ok(Header::Authority(value))
}
b"method" => {
@@ -68,11 +72,11 @@ impl Header {
Ok(Header::Method(method))
}
b"scheme" => {
let value = String::try_from(value)?;
let value = BytesStr::try_from(value)?;
Ok(Header::Scheme(value))
}
b"path" => {
let value = String::try_from(value)?;
let value = BytesStr::try_from(value)?;
Ok(Header::Path(value))
}
b"status" => {
@@ -213,10 +217,10 @@ impl<'a> Name<'a> {
name: name.clone(),
value: HeaderValue::from_bytes(&*value)?,
}),
Name::Authority => Ok(Header::Authority(String::try_from(value)?)),
Name::Authority => Ok(Header::Authority(BytesStr::try_from(value)?)),
Name::Method => Ok(Header::Method(Method::from_bytes(&*value)?)),
Name::Scheme => Ok(Header::Scheme(String::try_from(value)?)),
Name::Path => Ok(Header::Path(String::try_from(value)?)),
Name::Scheme => Ok(Header::Scheme(BytesStr::try_from(value)?)),
Name::Path => Ok(Header::Path(BytesStr::try_from(value)?)),
Name::Status => {
match StatusCode::from_bytes(&value) {
Ok(status) => Ok(Header::Status(status)),
@@ -238,3 +242,45 @@ impl<'a> Name<'a> {
}
}
}
// ===== impl BytesStr =====
impl BytesStr {
pub(crate) unsafe fn from_utf8_unchecked(bytes: Bytes) -> Self {
BytesStr(bytes)
}
#[doc(hidden)]
pub fn try_from(bytes: Bytes) -> Result<Self, std::str::Utf8Error> {
std::str::from_utf8(bytes.as_ref())?;
Ok(BytesStr(bytes))
}
pub(crate) fn as_str(&self) -> &str {
// Safety: check valid utf-8 in constructor
unsafe { std::str::from_utf8_unchecked(self.0.as_ref()) }
}
pub(crate) fn into_inner(self) -> Bytes {
self.0
}
}
impl std::ops::Deref for BytesStr {
type Target = str;
fn deref(&self) -> &str {
self.as_str()
}
}
impl AsRef<[u8]> for BytesStr {
fn as_ref(&self) -> &[u8] {
self.0.as_ref()
}
}
impl fmt::Debug for BytesStr {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.0.fmt(f)
}
}

2
src/hpack/huffman/mod.rs
View File

@@ -37,7 +37,7 @@ pub fn decode(src: &[u8], buf: &mut BytesMut) -> Result<BytesMut, DecoderError>
return Err(DecoderError::InvalidHuffmanCode);
}
Ok(buf.take())
Ok(buf.split())
}
// TODO: return error when there is not enough room to encode the value

2
src/hpack/mod.rs
View File

@@ -9,4 +9,4 @@ mod test;
pub use self::decoder::{Decoder, DecoderError, NeedMore};
pub use self::encoder::{Encode, EncodeState, Encoder, EncoderError};
pub use self::header::Header;
pub use self::header::{BytesStr, Header};

15
src/hpack/test/fixture.rs
View File

@@ -1,6 +1,6 @@
use crate::hpack::{Decoder, Encoder, Header};
use bytes::BytesMut;
use bytes::{buf::BufMutExt, BytesMut};
use hex::FromHex;
use serde_json::Value;
@@ -71,8 +71,10 @@ fn test_story(story: Value) {
decoder.queue_size_update(size);
}
let mut buf = BytesMut::with_capacity(case.wire.len());
buf.extend_from_slice(&case.wire);
decoder
.decode(&mut Cursor::new(&mut case.wire.clone().into()), |e| {
.decode(&mut Cursor::new(&mut buf), |e| {
let (name, value) = expect.remove(0);
assert_eq!(name, key_str(&e));
assert_eq!(value, value_str(&e));
@@ -87,7 +89,8 @@ fn test_story(story: Value) {
// Now, encode the headers
for case in &cases {
let mut buf = BytesMut::with_capacity(64 * 1024);
let limit = 64 * 1024;
let mut buf = BytesMut::with_capacity(limit);
if let Some(size) = case.header_table_size {
encoder.update_max_size(size);
@@ -104,7 +107,11 @@ fn test_story(story: Value) {
})
.collect();
encoder.encode(None, &mut input.clone().into_iter(), &mut buf);
encoder.encode(
None,
&mut input.clone().into_iter(),
&mut (&mut buf).limit(limit),
);
decoder
.decode(&mut Cursor::new(&mut buf), |e| {

27
src/hpack/test/fuzz.rs
View File

@@ -2,12 +2,13 @@ use crate::hpack::{Decoder, Encode, Encoder, Header};
use http::header::{HeaderName, HeaderValue};
use bytes::{Bytes, BytesMut};
use bytes::{buf::BufMutExt, Bytes, BytesMut};
use quickcheck::{Arbitrary, Gen, QuickCheck, TestResult};
use rand::{Rng, SeedableRng, StdRng};
use std::io::Cursor;
const MIN_CHUNK: usize = 16;
const MAX_CHUNK: usize = 2 * 1024;
#[test]
@@ -23,6 +24,16 @@ fn hpack_fuzz() {
.quickcheck(prop as fn(FuzzHpack) -> TestResult)
}
/*
// If wanting to test with a specific feed, uncomment and fill in the seed.
#[test]
fn hpack_fuzz_seeded() {
let _ = env_logger::try_init();
let seed = [/* fill me in*/];
FuzzHpack::new(seed).run();
}
*/
#[derive(Debug, Clone)]
struct FuzzHpack {
// The magic seed that makes the test case reproducible
@@ -121,7 +132,7 @@ impl FuzzHpack {
let mut chunks = vec![];
for _ in 0..rng.gen_range(0, 100) {
chunks.push(rng.gen_range(0, MAX_CHUNK));
chunks.push(rng.gen_range(MIN_CHUNK, MAX_CHUNK));
}
FuzzHpack {
@@ -165,7 +176,8 @@ impl FuzzHpack {
let mut input = frame.headers.into_iter();
let mut index = None;
let mut buf = BytesMut::with_capacity(chunks.pop().unwrap_or(MAX_CHUNK));
let mut max_chunk = chunks.pop().unwrap_or(MAX_CHUNK);
let mut buf = BytesMut::with_capacity(max_chunk);
if let Some(max) = frame.resizes.iter().max() {
decoder.queue_size_update(*max);
@@ -177,7 +189,7 @@ impl FuzzHpack {
}
loop {
match encoder.encode(index.take(), &mut input, &mut buf) {
match encoder.encode(index.take(), &mut input, &mut (&mut buf).limit(max_chunk)) {
Encode::Full => break,
Encode::Partial(i) => {
index = Some(i);
@@ -190,7 +202,8 @@ impl FuzzHpack {
})
.expect("partial decode");
buf = BytesMut::with_capacity(chunks.pop().unwrap_or(MAX_CHUNK));
max_chunk = chunks.pop().unwrap_or(MAX_CHUNK);
buf = BytesMut::with_capacity(max_chunk);
}
}
}
@@ -390,7 +403,7 @@ fn gen_string(g: &mut StdRng, min: usize, max: usize) -> String {
String::from_utf8(bytes).unwrap()
}
fn to_shared(src: String) -> ::string::String<Bytes> {
fn to_shared(src: String) -> crate::hpack::BytesStr {
let b: Bytes = src.into();
unsafe { ::string::String::from_utf8_unchecked(b) }
unsafe { crate::hpack::BytesStr::from_utf8_unchecked(b) }
}

26
src/proto/connection.rs
View File

@@ -5,18 +5,18 @@ use crate::{client, frame, proto, server};
use crate::frame::DEFAULT_INITIAL_WINDOW_SIZE;
use crate::proto::*;
use bytes::{Bytes, IntoBuf};
use bytes::{Buf, Bytes};
use futures_core::Stream;
use std::io;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio::io::{AsyncRead, AsyncWrite};
/// An H2 connection
#[derive(Debug)]
pub(crate) struct Connection<T, P, B: IntoBuf = Bytes>
pub(crate) struct Connection<T, P, B: Buf = Bytes>
where
P: Peer,
{
@@ -30,7 +30,7 @@ where
error: Option<Reason>,
/// Read / write frame values
codec: Codec<T, Prioritized<B::Buf>>,
codec: Codec<T, Prioritized<B>>,
/// Pending GOAWAY frames to write.
go_away: GoAway,
@@ -42,7 +42,7 @@ where
settings: Settings,
/// Stream state handler
streams: Streams<B::Buf, P>,
streams: Streams<B, P>,
/// Client or server
_phantom: PhantomData<P>,
@@ -73,10 +73,9 @@ impl<T, P, B> Connection<T, P, B>
where
T: AsyncRead + AsyncWrite + Unpin,
P: Peer,
B: IntoBuf + Unpin,
B::Buf: Unpin,
B: Buf + Unpin,
{
pub fn new(codec: Codec<T, Prioritized<B::Buf>>, config: Config) -> Connection<T, P, B> {
pub fn new(codec: Codec<T, Prioritized<B>>, config: Config) -> Connection<T, P, B> {
let streams = Streams::new(streams::Config {
local_init_window_sz: config
.settings
@@ -385,9 +384,9 @@ where
impl<T, B> Connection<T, client::Peer, B>
where
T: AsyncRead + AsyncWrite,
B: IntoBuf,
B: Buf,
{
pub(crate) fn streams(&self) -> &Streams<B::Buf, client::Peer> {
pub(crate) fn streams(&self) -> &Streams<B, client::Peer> {
&self.streams
}
}
@@ -395,10 +394,9 @@ where
impl<T, B> Connection<T, server::Peer, B>
where
T: AsyncRead + AsyncWrite + Unpin,
B: IntoBuf + Unpin,
B::Buf: Unpin,
B: Buf + Unpin,
{
pub fn next_incoming(&mut self) -> Option<StreamRef<B::Buf>> {
pub fn next_incoming(&mut self) -> Option<StreamRef<B>> {
self.streams.next_incoming()
}
@@ -431,7 +429,7 @@ where
impl<T, P, B> Drop for Connection<T, P, B>
where
P: Peer,
B: IntoBuf,
B: Buf,
{
fn drop(&mut self) {
// Ignore errors as this indicates that the mutex is poisoned.

2
src/proto/go_away.rs
View File

@@ -4,7 +4,7 @@ use crate::frame::{self, Reason, StreamId};
use bytes::Buf;
use std::io;
use std::task::{Context, Poll};
use tokio_io::AsyncWrite;
use tokio::io::AsyncWrite;
/// Manages our sending of GOAWAY frames.
#[derive(Debug)]

2
src/proto/mod.rs
View File

@@ -23,7 +23,7 @@ use crate::frame::{self, Frame};
use bytes::Buf;
use tokio_io::AsyncWrite;
use tokio::io::AsyncWrite;
pub type PingPayload = [u8; 8];

8
src/proto/ping_pong.rs
View File

@@ -3,12 +3,12 @@ use crate::frame::Ping;
use crate::proto::{self, PingPayload};
use bytes::Buf;
use futures_util::task::AtomicWaker;
use std::io;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};
use tokio_io::AsyncWrite;
use tokio_sync::AtomicWaker;
use tokio::io::AsyncWrite;
/// Acknowledges ping requests from the remote.
#[derive(Debug)]
@@ -190,7 +190,7 @@ impl PingPong {
.state
.store(USER_STATE_PENDING_PONG, Ordering::Release);
} else {
users.0.ping_task.register_by_ref(cx.waker());
users.0.ping_task.register(cx.waker());
}
}
@@ -233,7 +233,7 @@ impl UserPings {
pub(crate) fn poll_pong(&self, cx: &mut Context) -> Poll<Result<(), proto::Error>> {
// Must register before checking state, in case state were to change
// before we could register, and then the ping would just be lost.
self.0.pong_task.register_by_ref(cx.waker());
self.0.pong_task.register(cx.waker());
let prev = self.0.state.compare_and_swap(
USER_STATE_RECEIVED_PONG, // current
USER_STATE_EMPTY, // new

2
src/proto/streams/prioritize.rs
View File

@@ -6,7 +6,7 @@ use crate::frame::{Reason, StreamId};
use crate::codec::UserError;
use crate::codec::UserError::*;
use bytes::buf::Take;
use bytes::buf::ext::{BufExt, Take};
use std::io;
use std::task::{Context, Poll, Waker};
use std::{cmp, fmt, mem};

2
src/proto/streams/send.rs
View File

@@ -8,7 +8,7 @@ use crate::frame::{self, Reason};
use bytes::Buf;
use http;
use std::task::{Context, Poll, Waker};
use tokio_io::AsyncWrite;
use tokio::io::AsyncWrite;
use std::io;

2
src/proto/streams/streams.rs
View File

@@ -9,7 +9,7 @@ use crate::{client, proto, server};
use bytes::{Buf, Bytes};
use http::{HeaderMap, Request, Response};
use std::task::{Context, Poll, Waker};
use tokio_io::AsyncWrite;
use tokio::io::AsyncWrite;
use crate::PollExt;
use std::sync::{Arc, Mutex};

100
src/server.rs
View File

@@ -120,14 +120,14 @@ use crate::frame::{self, Pseudo, PushPromiseHeaderError, Reason, Settings, Strea
use crate::proto::{self, Config, Prioritized};
use crate::{FlowControl, PingPong, RecvStream, SendStream};
use bytes::{Buf, Bytes, IntoBuf};
use bytes::{Buf, Bytes};
use http::{HeaderMap, Request, Response};
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use std::{convert, fmt, io, mem};
use tokio_io::{AsyncRead, AsyncWrite};
use tokio::io::{AsyncRead, AsyncWrite};
/// In progress HTTP/2.0 connection handshake future.
///
@@ -144,7 +144,7 @@ use tokio_io::{AsyncRead, AsyncWrite};
///
/// [module]: index.html
#[must_use = "futures do nothing unless polled"]
pub struct Handshake<T, B: IntoBuf = Bytes> {
pub struct Handshake<T, B: Buf = Bytes> {
/// The config to pass to Connection::new after handshake succeeds.
builder: Builder,
/// The current state of the handshake.
@@ -172,7 +172,7 @@ pub struct Handshake<T, B: IntoBuf = Bytes> {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server;
/// # use h2::server::*;
/// #
@@ -188,7 +188,7 @@ pub struct Handshake<T, B: IntoBuf = Bytes> {
/// # pub fn main() {}
/// ```
#[must_use = "streams do nothing unless polled"]
pub struct Connection<T, B: IntoBuf> {
pub struct Connection<T, B: Buf> {
connection: proto::Connection<T, Peer, B>,
}
@@ -210,7 +210,7 @@ pub struct Connection<T, B: IntoBuf> {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server::*;
/// #
/// # fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
@@ -258,8 +258,8 @@ pub struct Builder {
///
/// [module]: index.html
#[derive(Debug)]
pub struct SendResponse<B: IntoBuf> {
inner: proto::StreamRef<B::Buf>,
pub struct SendResponse<B: Buf> {
inner: proto::StreamRef<B>,
}
/// Send a response to a promised request
@@ -276,26 +276,23 @@ pub struct SendResponse<B: IntoBuf> {
/// See [module] level docs for more details.
///
/// [module]: index.html
pub struct SendPushedResponse<B: IntoBuf> {
pub struct SendPushedResponse<B: Buf> {
inner: SendResponse<B>,
}
// Manual implementation necessary because of rust-lang/rust#26925
impl<B: IntoBuf + fmt::Debug> fmt::Debug for SendPushedResponse<B>
where
<B as bytes::IntoBuf>::Buf: std::fmt::Debug,
{
impl<B: Buf + fmt::Debug> fmt::Debug for SendPushedResponse<B> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "SendPushedResponse {{ {:?} }}", self.inner)
}
}
/// Stages of an in-progress handshake.
enum Handshaking<T, B: IntoBuf> {
enum Handshaking<T, B: Buf> {
/// State 1. Connection is flushing pending SETTINGS frame.
Flushing(Flush<T, Prioritized<B::Buf>>),
Flushing(Flush<T, Prioritized<B>>),
/// State 2. Connection is waiting for the client preface.
ReadingPreface(ReadPreface<T, Prioritized<B::Buf>>),
ReadingPreface(ReadPreface<T, Prioritized<B>>),
/// Dummy state for `mem::replace`.
Empty,
}
@@ -334,7 +331,7 @@ const PREFACE: [u8; 24] = *b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server;
/// # use h2::server::*;
/// #
@@ -359,8 +356,7 @@ where
impl<T, B> Connection<T, B>
where
T: AsyncRead + AsyncWrite + Unpin,
B: IntoBuf + Unpin,
B::Buf: Unpin + 'static,
B: Buf + Unpin + 'static,
{
fn handshake2(io: T, builder: Builder) -> Handshake<T, B> {
// Create the codec.
@@ -527,8 +523,7 @@ where
impl<T, B> futures_core::Stream for Connection<T, B>
where
T: AsyncRead + AsyncWrite + Unpin,
B: IntoBuf + Unpin,
B::Buf: Unpin + 'static,
B: Buf + Unpin + 'static,
{
type Item = Result<(Request<RecvStream>, SendResponse<B>), crate::Error>;
@@ -540,8 +535,7 @@ where
impl<T, B> fmt::Debug for Connection<T, B>
where
T: fmt::Debug,
B: fmt::Debug + IntoBuf,
B::Buf: fmt::Debug,
B: fmt::Debug + Buf,
{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_struct("Connection")
@@ -561,7 +555,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server::*;
/// #
/// # fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
@@ -600,7 +594,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server::*;
/// #
/// # fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
@@ -634,7 +628,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server::*;
/// #
/// # fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
@@ -667,7 +661,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server::*;
/// #
/// # fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
@@ -706,7 +700,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server::*;
/// #
/// # fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
@@ -754,7 +748,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server::*;
/// #
/// # fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
@@ -800,7 +794,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server::*;
/// #
/// # fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
@@ -846,7 +840,7 @@ impl Builder {
/// # Examples
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server::*;
/// # use std::time::Duration;
/// #
@@ -889,7 +883,7 @@ impl Builder {
/// Basic usage:
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server::*;
/// #
/// # fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
@@ -909,7 +903,7 @@ impl Builder {
/// type will be `&'static [u8]`.
///
/// ```
/// # use tokio_io::*;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # use h2::server::*;
/// #
/// # fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
@@ -927,8 +921,7 @@ impl Builder {
pub fn handshake<T, B>(&self, io: T) -> Handshake<T, B>
where
T: AsyncRead + AsyncWrite + Unpin,
B: IntoBuf + Unpin,
B::Buf: Unpin + 'static,
B: Buf + Unpin + 'static,
{
Connection::handshake2(io, self.clone())
}
@@ -942,7 +935,7 @@ impl Default for Builder {
// ===== impl SendResponse =====
impl<B: IntoBuf> SendResponse<B> {
impl<B: Buf> SendResponse<B> {
/// Send a response to a client request.
///
/// On success, a [`SendStream`] instance is returned. This instance can be
@@ -1034,7 +1027,7 @@ impl<B: IntoBuf> SendResponse<B> {
// ===== impl SendPushedResponse =====
impl<B: IntoBuf> SendPushedResponse<B> {
impl<B: Buf> SendPushedResponse<B> {
/// Send a response to a promised request.
///
/// On success, a [`SendStream`] instance is returned. This instance can be
@@ -1178,11 +1171,10 @@ where
// ===== impl Handshake =====
impl<T, B: IntoBuf> Future for Handshake<T, B>
impl<T, B: Buf> Future for Handshake<T, B>
where
T: AsyncRead + AsyncWrite + Unpin,
B: IntoBuf + Unpin,
B::Buf: Unpin + 'static,
B: Buf + Unpin + 'static,
{
type Output = Result<Connection<T, B>, crate::Error>;
@@ -1250,7 +1242,7 @@ where
impl<T, B> fmt::Debug for Handshake<T, B>
where
T: AsyncRead + AsyncWrite + fmt::Debug,
B: fmt::Debug + IntoBuf,
B: fmt::Debug + Buf,
{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
write!(fmt, "server::Handshake")
@@ -1363,10 +1355,10 @@ impl proto::Peer for Peer {
}}
};
b.version(Version::HTTP_2);
b = b.version(Version::HTTP_2);
if let Some(method) = pseudo.method {
b.method(method);
b = b.method(method);
} else {
malformed!("malformed headers: missing method");
}
@@ -1426,7 +1418,7 @@ impl proto::Peer for Peer {
})?);
}
b.uri(parts);
b = b.uri(parts);
let mut request = match b.body(()) {
Ok(request) => request,
@@ -1451,7 +1443,7 @@ impl proto::Peer for Peer {
impl<T, B> fmt::Debug for Handshaking<T, B>
where
B: IntoBuf,
B: Buf,
{
#[inline]
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
@@ -1463,35 +1455,35 @@ where
}
}
impl<T, B> convert::From<Flush<T, Prioritized<B::Buf>>> for Handshaking<T, B>
impl<T, B> convert::From<Flush<T, Prioritized<B>>> for Handshaking<T, B>
where
T: AsyncRead + AsyncWrite,
B: IntoBuf,
B: Buf,
{
#[inline]
fn from(flush: Flush<T, Prioritized<B::Buf>>) -> Self {
fn from(flush: Flush<T, Prioritized<B>>) -> Self {
Handshaking::Flushing(flush)
}
}
impl<T, B> convert::From<ReadPreface<T, Prioritized<B::Buf>>> for Handshaking<T, B>
impl<T, B> convert::From<ReadPreface<T, Prioritized<B>>> for Handshaking<T, B>
where
T: AsyncRead + AsyncWrite,
B: IntoBuf,
B: Buf,
{
#[inline]
fn from(read: ReadPreface<T, Prioritized<B::Buf>>) -> Self {
fn from(read: ReadPreface<T, Prioritized<B>>) -> Self {
Handshaking::ReadingPreface(read)
}
}
impl<T, B> convert::From<Codec<T, Prioritized<B::Buf>>> for Handshaking<T, B>
impl<T, B> convert::From<Codec<T, Prioritized<B>>> for Handshaking<T, B>
where
T: AsyncRead + AsyncWrite,
B: IntoBuf,
B: Buf,
{
#[inline]
fn from(codec: Codec<T, Prioritized<B::Buf>>) -> Self {
fn from(codec: Codec<T, Prioritized<B>>) -> Self {
Handshaking::from(Flush::new(codec))
}
}

12
src/share.rs
View File

@@ -2,7 +2,7 @@ use crate::codec::UserError;
use crate::frame::Reason;
use crate::proto::{self, WindowSize};
use bytes::{Bytes, IntoBuf};
use bytes::{Buf, Bytes};
use http::HeaderMap;
use crate::PollExt;
@@ -95,8 +95,8 @@ use std::task::{Context, Poll};
/// [`send_trailers`]: #method.send_trailers
/// [`send_reset`]: #method.send_reset
#[derive(Debug)]
pub struct SendStream<B: IntoBuf> {
inner: proto::StreamRef<B::Buf>,
pub struct SendStream<B: Buf> {
inner: proto::StreamRef<B>,
}
/// A stream identifier, as described in [Section 5.1.1] of RFC 7540.
@@ -219,8 +219,8 @@ pub struct Pong {
// ===== impl SendStream =====
impl<B: IntoBuf> SendStream<B> {
pub(crate) fn new(inner: proto::StreamRef<B::Buf>) -> Self {
impl<B: Buf> SendStream<B> {
pub(crate) fn new(inner: proto::StreamRef<B>) -> Self {
SendStream { inner }
}
@@ -333,7 +333,7 @@ impl<B: IntoBuf> SendStream<B> {
/// [`Error`]: struct.Error.html
pub fn send_data(&mut self, data: B, end_of_stream: bool) -> Result<(), crate::Error> {
self.inner
.send_data(data.into_buf(), end_of_stream)
.send_data(data, end_of_stream)
.map_err(Into::into)
}

6
tests/h2-fuzz/Cargo.toml
View File

@@ -9,7 +9,7 @@ edition = "2018"
h2 = { path = "../.." }
env_logger = { version = "0.5.3", default-features = false }
futures-preview = "=0.3.0-alpha.19"
futures = { version = "0.3", default-features = false }
honggfuzz = "0.5"
http = "0.1.3"
tokio = "=0.2.0-alpha.6"
http = { git = "https://github.com/hyperium/http" } #"0.1.3"
tokio = { version = "0.2", features = [] }

10
tests/h2-support/Cargo.toml
View File

@@ -7,9 +7,9 @@ edition = "2018"
[dependencies]
h2 = { path = "../..", features = ["unstable-stream", "unstable"] }
bytes = "0.4.7"
bytes = "0.5"
env_logger = "0.5.9"
futures-preview = "=0.3.0-alpha.19"
http = "0.1.5"
string = "0.2"
tokio = "=0.2.0-alpha.6"
futures = { version = "0.3", default-features = false }
http = { git = "https://github.com/hyperium/http" } #"0.1.3"
tokio = { version = "0.2", features = ["time"] }
tokio-test = "0.2"

5
tests/h2-support/src/client_ext.rs
View File

@@ -1,4 +1,4 @@
use bytes::IntoBuf;
use bytes::Buf;
use h2::client::{ResponseFuture, SendRequest};
use http::Request;
@@ -11,8 +11,7 @@ pub trait SendRequestExt {
impl<B> SendRequestExt for SendRequest<B>
where
B: IntoBuf + Unpin,
B::Buf: Unpin + 'static,
B: Buf + Unpin + 'static,
{
fn get(&mut self, uri: &str) -> ResponseFuture {
let req = Request::builder()

109
tests/h2-support/src/frames.rs
View File

@@ -1,9 +1,9 @@
use std::convert::TryInto;
use std::fmt;
use bytes::{Bytes, IntoBuf};
use http::{self, HeaderMap, HttpTryFrom};
use bytes::Bytes;
use http::{self, HeaderMap};
use super::SendFrame;
use h2::frame::{self, Frame, StreamId};
pub const SETTINGS: &'static [u8] = &[0, 0, 0, 4, 0, 0, 0, 0, 0];
@@ -25,9 +25,10 @@ where
pub fn data<T, B>(id: T, buf: B) -> Mock<frame::Data>
where
T: Into<StreamId>,
B: Into<Bytes>,
B: AsRef<[u8]>,
{
Mock(frame::Data::new(id.into(), buf.into()))
let buf = Bytes::copy_from_slice(buf.as_ref());
Mock(frame::Data::new(id.into(), buf))
}
pub fn push_promise<T1, T2>(id: T1, promised: T2) -> Mock<frame::PushPromise>
@@ -100,8 +101,10 @@ where
impl Mock<frame::Headers> {
pub fn request<M, U>(self, method: M, uri: U) -> Self
where
M: HttpTryInto<http::Method>,
U: HttpTryInto<http::Uri>,
M: TryInto<http::Method>,
M::Error: fmt::Debug,
U: TryInto<http::Uri>,
U::Error: fmt::Debug,
{
let method = method.try_into().unwrap();
let uri = uri.try_into().unwrap();
@@ -112,7 +115,8 @@ impl Mock<frame::Headers> {
pub fn response<S>(self, status: S) -> Self
where
S: HttpTryInto<http::StatusCode>,
S: TryInto<http::StatusCode>,
S::Error: fmt::Debug,
{
let status = status.try_into().unwrap();
let (id, _, fields) = self.into_parts();
@@ -128,8 +132,10 @@ impl Mock<frame::Headers> {
pub fn field<K, V>(self, key: K, value: V) -> Self
where
K: HttpTryInto<http::header::HeaderName>,
V: HttpTryInto<http::header::HeaderValue>,
K: TryInto<http::header::HeaderName>,
K::Error: fmt::Debug,
V: TryInto<http::header::HeaderValue>,
V::Error: fmt::Debug,
{
let (id, pseudo, mut fields) = self.into_parts();
fields.insert(key.try_into().unwrap(), value.try_into().unwrap());
@@ -170,12 +176,6 @@ impl From<Mock<frame::Headers>> for frame::Headers {
}
}
impl From<Mock<frame::Headers>> for SendFrame {
fn from(src: Mock<frame::Headers>) -> Self {
Frame::Headers(src.0)
}
}
// Data helpers
impl Mock<frame::Data> {
@@ -190,28 +190,15 @@ impl Mock<frame::Data> {
}
}
impl From<Mock<frame::Data>> for SendFrame {
fn from(src: Mock<frame::Data>) -> Self {
let id = src.0.stream_id();
let eos = src.0.is_end_stream();
let is_padded = src.0.is_padded();
let payload = src.0.into_payload();
let mut frame = frame::Data::new(id, payload.into_buf());
frame.set_end_stream(eos);
if is_padded {
frame.set_padded();
}
Frame::Data(frame)
}
}
// PushPromise helpers
impl Mock<frame::PushPromise> {
pub fn request<M, U>(self, method: M, uri: U) -> Self
where
M: HttpTryInto<http::Method>,
U: HttpTryInto<http::Uri>,
M: TryInto<http::Method>,
M::Error: fmt::Debug,
U: TryInto<http::Uri>,
U::Error: fmt::Debug,
{
let method = method.try_into().unwrap();
let uri = uri.try_into().unwrap();
@@ -229,8 +216,10 @@ impl Mock<frame::PushPromise> {
pub fn field<K, V>(self, key: K, value: V) -> Self
where
K: HttpTryInto<http::header::HeaderName>,
V: HttpTryInto<http::header::HeaderValue>,
K: TryInto<http::header::HeaderName>,
K::Error: fmt::Debug,
V: TryInto<http::header::HeaderValue>,
V::Error: fmt::Debug,
{
let (id, promised, pseudo, mut fields) = self.into_parts();
fields.insert(key.try_into().unwrap(), value.try_into().unwrap());
@@ -247,12 +236,6 @@ impl Mock<frame::PushPromise> {
}
}
impl From<Mock<frame::PushPromise>> for SendFrame {
fn from(src: Mock<frame::PushPromise>) -> Self {
Frame::PushPromise(src.0)
}
}
// GoAway helpers
impl Mock<frame::GoAway> {
@@ -281,12 +264,6 @@ impl Mock<frame::GoAway> {
}
}
impl From<Mock<frame::GoAway>> for SendFrame {
fn from(src: Mock<frame::GoAway>) -> Self {
Frame::GoAway(src.0)
}
}
// ==== Reset helpers
impl Mock<frame::Reset> {
@@ -326,12 +303,6 @@ impl Mock<frame::Reset> {
}
}
impl From<Mock<frame::Reset>> for SendFrame {
fn from(src: Mock<frame::Reset>) -> Self {
Frame::Reset(src.0)
}
}
// ==== Settings helpers
impl Mock<frame::Settings> {
@@ -357,12 +328,6 @@ impl From<Mock<frame::Settings>> for frame::Settings {
}
}
impl From<Mock<frame::Settings>> for SendFrame {
fn from(src: Mock<frame::Settings>) -> Self {
Frame::Settings(src.0)
}
}
// ==== Ping helpers
impl Mock<frame::Ping> {
@@ -371,29 +336,3 @@ impl Mock<frame::Ping> {
Mock(frame::Ping::pong(payload))
}
}
impl From<Mock<frame::Ping>> for SendFrame {
fn from(src: Mock<frame::Ping>) -> Self {
Frame::Ping(src.0)
}
}
// ==== "trait alias" for types that are HttpTryFrom and have Debug Errors ====
pub trait HttpTryInto<T> {
type Error: fmt::Debug;
fn try_into(self) -> Result<T, Self::Error>;
}
impl<T, U> HttpTryInto<T> for U
where
T: HttpTryFrom<U>,
T::Error: fmt::Debug,
{
type Error = T::Error;
fn try_into(self) -> Result<T, Self::Error> {
T::try_from(self)
}
}

5
tests/h2-support/src/lib.rs
View File

@@ -7,7 +7,6 @@ pub mod raw;
pub mod frames;
pub mod mock;
pub mod mock_io;
pub mod prelude;
pub mod util;
@@ -21,7 +20,7 @@ pub type WindowSize = usize;
pub const DEFAULT_WINDOW_SIZE: WindowSize = (1 << 16) - 1;
// This is our test Codec type
pub type Codec<T> = h2::Codec<T, ::std::io::Cursor<::bytes::Bytes>>;
pub type Codec<T> = h2::Codec<T, bytes::Bytes>;
// This is the frame type that is sent
pub type SendFrame = h2::frame::Frame<::std::io::Cursor<::bytes::Bytes>>;
pub type SendFrame = h2::frame::Frame<bytes::Bytes>;

25
tests/h2-support/src/mock.rs
View File

@@ -9,7 +9,6 @@ use futures::{ready, Stream, StreamExt};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use super::assert::assert_frame_eq;
use futures::executor::block_on;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
@@ -324,20 +323,18 @@ impl AsyncWrite for Handle {
impl Drop for Handle {
fn drop(&mut self) {
block_on(async {
poll_fn(|cx| {
assert!(self.codec.shutdown(cx).is_ready());
// Shutdown *shouldn't* need a real Waker...
let waker = futures::task::noop_waker();
let mut cx = Context::from_waker(&waker);
assert!(self.codec.shutdown(&mut cx).is_ready());
let mut me = self.codec.get_mut().inner.lock().unwrap();
me.closed = true;
if let Ok(mut me) = self.codec.get_mut().inner.lock() {
me.closed = true;
if let Some(task) = me.rx_task.take() {
task.wake();
}
Poll::Ready(())
})
.await;
});
if let Some(task) = me.rx_task.take() {
task.wake();
}
}
}
}
@@ -482,5 +479,5 @@ impl AsyncWrite for Pipe {
}
pub async fn idle_ms(ms: u64) {
tokio::timer::delay(tokio::clock::now() + Duration::from_millis(ms)).await
tokio::time::delay_for(Duration::from_millis(ms)).await
}

509
tests/h2-support/src/mock_io.rs
View File

@@ -1,509 +0,0 @@
//! A mock type implementing [`Read`] and [`Write`].
//!
//! Copied from https://github.com/carllerche/mock-io.
//!
//! TODO:
//! - Either the mock-io crate should be released or this module should be
//! removed from h2.
//!
//! # Overview
//!
//! Provides a type that implements [`Read`] + [`Write`] that can be configured
//! to handle an arbitrary sequence of read and write operations. This is useful
//! for writing unit tests for networking services as using an actual network
//! type is fairly non deterministic.
//!
//! # Usage
//!
//! Add the following to your `Cargo.toml`
//!
//! ```toml
//! [dependencies]
//! mock-io = { git = "https://github.com/carllerche/mock-io" }
//! ```
//!
//! Then use it in your project. For example, a test could be written:
//!
//! ```
//! use mock_io::{Builder, Mock};
//! use std::io::{Read, Write};
//!
//! # /*
//! #[test]
//! # */
//! fn test_io() {
//! let mut mock = Builder::new()
//! .write(b"ping")
//! .read(b"pong")
//! .build();
//!
//! let n = mock.write(b"ping").unwrap();
//! assert_eq!(n, 4);
//!
//! let mut buf = vec![];
//! mock.read_to_end(&mut buf).unwrap();
//!
//! assert_eq!(buf, b"pong");
//! }
//! # pub fn main() {
//! # test_io();
//! # }
//! ```
//!
//! Attempting to write data that the mock isn't expected will result in a
//! panic.
//!
//! # Tokio
//!
//! `Mock` also supports tokio by implementing `AsyncRead` and `AsyncWrite`.
//! When using `Mock` in context of a Tokio task, it will automatically switch
//! to "async" behavior (this can also be set explicitly by calling `set_async`
//! on `Builder`).
//!
//! In async mode, calls to read and write are non-blocking and the task using
//! the mock is notified when the readiness state changes.
//!
//! # `io-dump` dump files
//!
//! `Mock` can also be configured from an `io-dump` file. By doing this, the
//! mock value will replay a previously recorded behavior. This is useful for
//! collecting a scenario from the real world and replying it as part of a test.
//!
//! [`Read`]: https://doc.rust-lang.org/std/io/trait.Read.html
//! [`Write`]: https://doc.rust-lang.org/std/io/trait.Write.html
#![allow(deprecated)]
use std::collections::VecDeque;
use std::time::{Duration, Instant};
use std::{cmp, io};
/// An I/O handle that follows a predefined script.
///
/// This value is created by `Builder` and implements `Read + `Write`. It
/// follows the scenario described by the builder and panics otherwise.
#[derive(Debug)]
pub struct Mock {
inner: Inner,
tokio: tokio_::Inner,
}
#[derive(Debug)]
pub struct Handle {
inner: tokio_::Handle,
}
/// Builds `Mock` instances.
#[derive(Debug, Clone, Default)]
pub struct Builder {
// Sequence of actions for the Mock to take
actions: VecDeque<Action>,
}
#[derive(Debug, Clone)]
enum Action {
Read(Vec<u8>),
Write(Vec<u8>),
Wait(Duration),
}
#[derive(Debug)]
struct Inner {
actions: VecDeque<Action>,
waiting: Option<Instant>,
}
impl Builder {
/// Return a new, empty `Builder.
pub fn new() -> Self {
Self::default()
}
/// Sequence a `read` operation.
///
/// The next operation in the mock's script will be to expect a `read` call
/// and return `buf`.
pub fn read(&mut self, buf: &[u8]) -> &mut Self {
self.actions.push_back(Action::Read(buf.into()));
self
}
/// Sequence a `write` operation.
///
/// The next operation in the mock's script will be to expect a `write`
/// call.
pub fn write(&mut self, buf: &[u8]) -> &mut Self {
self.actions.push_back(Action::Write(buf.into()));
self
}
/// Sequence a wait.
///
/// The next operation in the mock's script will be to wait without doing so
/// for `duration` amount of time.
pub fn wait(&mut self, duration: Duration) -> &mut Self {
let duration = cmp::max(duration, Duration::from_millis(1));
self.actions.push_back(Action::Wait(duration));
self
}
/// Build a `Mock` value according to the defined script.
pub fn build(&mut self) -> Mock {
let (mock, _) = self.build_with_handle();
mock
}
/// Build a `Mock` value paired with a handle
pub fn build_with_handle(&mut self) -> (Mock, Handle) {
let (tokio, handle) = tokio_::Inner::new();
let src = self.clone();
let mock = Mock {
inner: Inner {
actions: src.actions,
waiting: None,
},
tokio: tokio,
};
let handle = Handle { inner: handle };
(mock, handle)
}
}
impl Handle {
/// Sequence a `read` operation.
///
/// The next operation in the mock's script will be to expect a `read` call
/// and return `buf`.
pub fn read(&mut self, buf: &[u8]) -> &mut Self {
self.inner.read(buf);
self
}
/// Sequence a `write` operation.
///
/// The next operation in the mock's script will be to expect a `write`
/// call.
pub fn write(&mut self, buf: &[u8]) -> &mut Self {
self.inner.write(buf);
self
}
}
impl Inner {
fn read(&mut self, dst: &mut [u8]) -> io::Result<usize> {
match self.action() {
Some(&mut Action::Read(ref mut data)) => {
// Figure out how much to copy
let n = cmp::min(dst.len(), data.len());
// Copy the data into the `dst` slice
(&mut dst[..n]).copy_from_slice(&data[..n]);
// Drain the data from the source
data.drain(..n);
// Return the number of bytes read
Ok(n)
}
Some(_) => {
// Either waiting or expecting a write
Err(io::ErrorKind::WouldBlock.into())
}
None => Ok(0),
}
}
fn write(&mut self, mut src: &[u8]) -> io::Result<usize> {
let mut ret = 0;
if self.actions.is_empty() {
return Err(io::ErrorKind::BrokenPipe.into());
}
match self.action() {
Some(&mut Action::Wait(..)) => {
return Err(io::ErrorKind::WouldBlock.into());
}
_ => {}
}
for i in 0..self.actions.len() {
match self.actions[i] {
Action::Write(ref mut expect) => {
let n = cmp::min(src.len(), expect.len());
assert_eq!(&src[..n], &expect[..n]);
// Drop data that was matched
expect.drain(..n);
src = &src[n..];
ret += n;
if src.is_empty() {
return Ok(ret);
}
}
Action::Wait(..) => {
break;
}
_ => {}
}
// TODO: remove write
}
Ok(ret)
}
fn remaining_wait(&mut self) -> Option<Duration> {
match self.action() {
Some(&mut Action::Wait(dur)) => Some(dur),
_ => None,
}
}
fn action(&mut self) -> Option<&mut Action> {
loop {
if self.actions.is_empty() {
return None;
}
match self.actions[0] {
Action::Read(ref mut data) => {
if !data.is_empty() {
break;
}
}
Action::Write(ref mut data) => {
if !data.is_empty() {
break;
}
}
Action::Wait(ref mut dur) => {
if let Some(until) = self.waiting {
let now = Instant::now();
if now < until {
break;
}
} else {
self.waiting = Some(Instant::now() + *dur);
break;
}
}
}
let _action = self.actions.pop_front();
}
self.actions.front_mut()
}
}
// use tokio::*;
mod tokio_ {
use super::*;
use futures::channel::mpsc;
use futures::{ready, FutureExt, Stream};
use std::task::{Context, Poll, Waker};
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::timer::Delay;
use std::pin::Pin;
use std::io;
#[derive(Debug)]
pub struct Inner {
sleep: Option<Delay>,
read_wait: Option<Waker>,
rx: mpsc::UnboundedReceiver<Action>,
}
#[derive(Debug)]
pub struct Handle {
tx: mpsc::UnboundedSender<Action>,
}
// ===== impl Handle =====
impl Handle {
pub fn read(&mut self, buf: &[u8]) {
self.tx.unbounded_send(Action::Read(buf.into())).unwrap();
}
pub fn write(&mut self, buf: &[u8]) {
self.tx.unbounded_send(Action::Write(buf.into())).unwrap();
}
}
// ===== impl Inner =====
impl Inner {
pub fn new() -> (Inner, Handle) {
let (tx, rx) = mpsc::unbounded();
let inner = Inner {
sleep: None,
read_wait: None,
rx: rx,
};
let handle = Handle { tx };
(inner, handle)
}
pub(super) fn poll_action(&mut self, cx: &mut Context) -> Poll<Option<Action>> {
Pin::new(&mut self.rx).poll_next(cx)
}
}
impl Mock {
fn maybe_wakeup_reader(&mut self) {
match self.inner.action() {
Some(&mut Action::Read(_)) | None => {
if let Some(task) = self.tokio.read_wait.take() {
task.wake();
}
}
_ => {}
}
}
}
impl AsyncRead for Mock {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<io::Result<usize>> {
loop {
if let Some(sleep) = &mut self.tokio.sleep {
ready!(sleep.poll_unpin(cx));
}
// If a sleep is set, it has already fired
self.tokio.sleep = None;
match self.inner.read(buf) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
if let Some(rem) = self.inner.remaining_wait() {
self.tokio.sleep = Some(tokio::timer::delay(Instant::now() + rem));
} else {
self.tokio.read_wait = Some(cx.waker().clone());
return Poll::Pending;
}
}
Ok(0) => {
// TODO: Extract
match self.tokio.poll_action(cx) {
Poll::Ready(Some(action)) => {
self.inner.actions.push_back(action);
continue;
}
Poll::Ready(None) => {
return Poll::Ready(Ok(0));
}
Poll::Pending => {
return Poll::Pending;
}
}
}
ret => return Poll::Ready(ret),
}
}
}
}
impl AsyncWrite for Mock {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<Result<usize, io::Error>> {
loop {
if let Some(sleep) = &mut self.tokio.sleep {
ready!(sleep.poll_unpin(cx));
}
// If a sleep is set, it has already fired
self.tokio.sleep = None;
match self.inner.write(buf) {
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
if let Some(rem) = self.inner.remaining_wait() {
self.tokio.sleep = Some(tokio::timer::delay(Instant::now() + rem));
} else {
panic!("unexpected WouldBlock");
}
}
Ok(0) => {
// TODO: Is this correct?
if !self.inner.actions.is_empty() {
return Poll::Pending;
}
// TODO: Extract
match self.tokio.poll_action(cx) {
Poll::Ready(Some(action)) => {
self.inner.actions.push_back(action);
continue;
}
Poll::Ready(None) => {
panic!("unexpected write");
}
Poll::Pending => return Poll::Pending,
}
}
ret => {
self.maybe_wakeup_reader();
return Poll::Ready(ret);
}
}
}
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Result<(), io::Error>> {
Poll::Ready(Ok(()))
}
}
/*
TODO: Is this required?
/// Returns `true` if called from the context of a futures-rs Task
pub fn is_task_ctx() -> bool {
use std::panic;
// Save the existing panic hook
let h = panic::take_hook();
// Install a new one that does nothing
panic::set_hook(Box::new(|_| {}));
// Attempt to call the fn
let r = panic::catch_unwind(|| task::current()).is_ok();
// Re-install the old one
panic::set_hook(h);
// Return the result
r
}
*/
}

12
tests/h2-support/src/prelude.rs
View File

@@ -27,7 +27,7 @@ pub use super::{
pub use super::assert::assert_frame_eq;
// Re-export useful crates
pub use super::mock_io;
pub use tokio_test::io as mock_io;
pub use {bytes, env_logger, futures, http, tokio::io as tokio_io};
// Re-export primary future types
@@ -42,7 +42,10 @@ pub use super::client_ext::SendRequestExt;
// Re-export HTTP types
pub use http::{uri, HeaderMap, Method, Request, Response, StatusCode, Version};
pub use bytes::{Buf, BufMut, Bytes, BytesMut, IntoBuf};
pub use bytes::{
buf::{BufExt, BufMutExt},
Buf, BufMut, Bytes, BytesMut,
};
pub use tokio::io::{AsyncRead, AsyncWrite};
@@ -61,7 +64,7 @@ pub trait MockH2 {
fn handshake(&mut self) -> &mut Self;
}
impl MockH2 for super::mock_io::Builder {
impl MockH2 for tokio_test::io::Builder {
fn handshake(&mut self) -> &mut Self {
self.write(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")
// Settings frame
@@ -81,8 +84,7 @@ pub trait ClientExt {
impl<T, B> ClientExt for client::Connection<T, B>
where
T: AsyncRead + AsyncWrite + Unpin + 'static,
B: IntoBuf + Unpin + 'static,
B::Buf: Unpin,
B: Buf + Unpin + 'static,
{
fn run<'a, F: Future + Unpin + 'a>(
&'a mut self,

2
tests/h2-support/src/raw.rs
View File

@@ -7,7 +7,7 @@ macro_rules! raw_codec {
$fn:ident => [$($chunk:expr,)+];
)*
) => {{
let mut b = $crate::mock_io::Builder::new();
let mut b = $crate::prelude::mock_io::Builder::new();
$({
let mut chunk = vec![];

15
tests/h2-support/src/util.rs
View File

@@ -1,14 +1,21 @@
use h2;
use bytes::Bytes;
use bytes::{BufMut, Bytes};
use futures::ready;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use string::{String, TryFrom};
pub fn byte_str(s: &str) -> String<Bytes> {
String::try_from(Bytes::from(s)).unwrap()
pub fn byte_str(s: &str) -> h2::frame::BytesStr {
h2::frame::BytesStr::try_from(Bytes::copy_from_slice(s.as_bytes())).unwrap()
}
pub async fn concat(mut body: h2::RecvStream) -> Result<Bytes, h2::Error> {
let mut vec = Vec::new();
while let Some(chunk) = body.data().await {
vec.put(chunk?);
}
Ok(vec.into())
}
pub async fn yield_once() {

4
tests/h2-tests/Cargo.toml
View File

@@ -10,5 +10,5 @@ edition = "2018"
[dev-dependencies]
h2-support = { path = "../h2-support" }
log = "0.4.1"
futures-preview = "=0.3.0-alpha.19"
tokio = "=0.2.0-alpha.6"
futures = { version = "0.3", default-features = false, features = ["alloc"] }
tokio = { version = "0.2", features = ["macros", "tcp"] }

22
tests/h2-tests/tests/client_request.rs
View File

@@ -418,7 +418,7 @@ async fn send_reset_notifies_recv_stream() {
// We don't want a join, since any of the other futures notifying
// will make the rx future polled again, but we are
// specifically testing that rx gets notified on its own.
let mut unordered = FuturesUnordered::<Pin<Box<dyn Future<Output = ()>>>>::new();
let unordered = FuturesUnordered::<Pin<Box<dyn Future<Output = ()>>>>::new();
unordered.push(Box::pin(rx));
unordered.push(Box::pin(tx));
@@ -754,7 +754,7 @@ async fn pending_send_request_gets_reset_by_peer_properly() {
let _ = env_logger::try_init();
let (io, mut srv) = mock::new();
let payload = vec![0; (frame::DEFAULT_INITIAL_WINDOW_SIZE * 2) as usize];
let payload = Bytes::from(vec![0; (frame::DEFAULT_INITIAL_WINDOW_SIZE * 2) as usize]);
let max_frame_size = frame::DEFAULT_MAX_FRAME_SIZE as usize;
let srv = async {
@@ -811,7 +811,7 @@ async fn pending_send_request_gets_reset_by_peer_properly() {
};
// Send the data
stream.send_data(payload[..].into(), true).unwrap();
stream.send_data(payload.clone(), true).unwrap();
conn.drive(response).await;
drop(client);
drop(stream);
@@ -897,7 +897,7 @@ async fn notify_on_send_capacity() {
// This test ensures that the client gets notified when there is additional
// send capacity. In other words, when the server is ready to accept a new
// stream, the client is notified.
use futures::channel::oneshot;
use tokio::sync::oneshot;
let _ = env_logger::try_init();
@@ -1016,13 +1016,14 @@ async fn send_stream_poll_reset() {
async fn drop_pending_open() {
// This test checks that a stream queued for pending open behaves correctly when its
// client drops.
use tokio::sync::oneshot;
let _ = env_logger::try_init();
let (io, mut srv) = mock::new();
let (init_tx, init_rx) = futures::channel::oneshot::channel();
let (trigger_go_away_tx, trigger_go_away_rx) = futures::channel::oneshot::channel();
let (sent_go_away_tx, sent_go_away_rx) = futures::channel::oneshot::channel();
let (drop_tx, drop_rx) = futures::channel::oneshot::channel();
let (init_tx, init_rx) = oneshot::channel();
let (trigger_go_away_tx, trigger_go_away_rx) = oneshot::channel();
let (sent_go_away_tx, sent_go_away_rx) = oneshot::channel();
let (drop_tx, drop_rx) = oneshot::channel();
let mut settings = frame::Settings::default();
settings.set_max_concurrent_streams(Some(2));
@@ -1103,11 +1104,12 @@ async fn malformed_response_headers_dont_unlink_stream() {
// This test checks that receiving malformed headers frame on a stream with
// no remaining references correctly resets the stream, without prematurely
// unlinking it.
use tokio::sync::oneshot;
let _ = env_logger::try_init();
let (io, mut srv) = mock::new();
let (drop_tx, drop_rx) = futures::channel::oneshot::channel();
let (queued_tx, queued_rx) = futures::channel::oneshot::channel();
let (drop_tx, drop_rx) = oneshot::channel();
let (queued_tx, queued_rx) = oneshot::channel();
let srv = async move {
let settings = srv.assert_client_handshake().await;

3
tests/h2-tests/tests/codec_read.rs
View File

@@ -175,8 +175,7 @@ async fn read_continuation_frames() {
let expected = large
.iter()
.fold(HeaderMap::new(), |mut map, &(name, ref value)| {
use h2_support::frames::HttpTryInto;
map.append(name, value.as_str().try_into().unwrap());
map.append(name, value.parse().unwrap());
map
});
assert_eq!(head.headers, expected);

4
tests/h2-tests/tests/codec_write.rs
View File

@@ -27,10 +27,10 @@ async fn write_continuation_frames() {
let (mut client, mut conn) = client::handshake(io).await.expect("handshake");
let mut request = Request::builder();
request.uri("https://http2.akamai.com/");
request = request.uri("https://http2.akamai.com/");
for &(name, ref value) in &large {
request.header(name, &value[..]);
request = request.header(name, &value[..]);
}
let request = request.body(()).unwrap();

56
tests/h2-tests/tests/flow_control.rs
View File

@@ -9,7 +9,7 @@ use h2_support::util::yield_once;
async fn send_data_without_requesting_capacity() {
let _ = env_logger::try_init();
let payload = [0; 1024];
let payload = vec![0; 1024];
let mock = mock_io::Builder::new()
.handshake()
@@ -42,7 +42,7 @@ async fn send_data_without_requesting_capacity() {
assert_eq!(stream.capacity(), 0);
// Send the data
stream.send_data(payload[..].into(), true).unwrap();
stream.send_data(payload.into(), true).unwrap();
// Get the response
let resp = h2.run(response).await.unwrap();
@@ -93,17 +93,17 @@ async fn release_capacity_sends_window_update() {
let mut body = resp.into_parts().1;
// read some body to use up window size to below half
let buf = body.next().await.unwrap().unwrap();
let buf = body.data().await.unwrap().unwrap();
assert_eq!(buf.len(), payload_len);
let buf = body.next().await.unwrap().unwrap();
let buf = body.data().await.unwrap().unwrap();
assert_eq!(buf.len(), payload_len);
let buf = body.next().await.unwrap().unwrap();
let buf = body.data().await.unwrap().unwrap();
assert_eq!(buf.len(), payload_len);
body.flow_control().release_capacity(buf.len() * 2).unwrap();
let buf = body.next().await.unwrap().unwrap();
let buf = body.data().await.unwrap().unwrap();
assert_eq!(buf.len(), payload_len);
};
@@ -153,11 +153,11 @@ async fn release_capacity_of_small_amount_does_not_send_window_update() {
assert_eq!(resp.status(), StatusCode::OK);
let mut body = resp.into_parts().1;
assert!(!body.is_end_stream());
let buf = body.next().await.unwrap().unwrap();
let buf = body.data().await.unwrap().unwrap();
// read the small body and then release it
assert_eq!(buf.len(), 16);
body.flow_control().release_capacity(buf.len()).unwrap();
let buf = body.next().await;
let buf = body.data().await;
assert!(buf.is_none());
};
join(async move { h2.await.unwrap() }, req).await;
@@ -213,7 +213,7 @@ async fn recv_data_overflows_connection_window() {
let resp = client.send_request(request, true).unwrap().0.await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = resp.into_parts().1;
let res = body.try_concat().await;
let res = util::concat(body).await;
let err = res.unwrap_err();
assert_eq!(
err.to_string(),
@@ -274,7 +274,7 @@ async fn recv_data_overflows_stream_window() {
let resp = client.send_request(request, true).unwrap().0.await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = resp.into_parts().1;
let res = body.try_concat().await;
let res = util::concat(body).await;
let err = res.unwrap_err();
assert_eq!(
err.to_string(),
@@ -685,8 +685,7 @@ async fn reserved_capacity_assigned_in_multi_window_updates() {
#[tokio::test]
async fn connection_notified_on_released_capacity() {
use futures::channel::mpsc;
use futures::channel::oneshot;
use tokio::sync::{mpsc, oneshot};
let _ = env_logger::try_init();
let (io, mut srv) = mock::new();
@@ -695,7 +694,7 @@ async fn connection_notified_on_released_capacity() {
// notifications. This test is here, in part, to ensure that the connection
// receives the appropriate notifications to send out window updates.
let (tx, mut rx) = mpsc::unbounded();
let (tx, mut rx) = mpsc::unbounded_channel();
// Because threading is fun
let (settings_tx, settings_rx) = oneshot::channel();
@@ -744,11 +743,11 @@ async fn connection_notified_on_released_capacity() {
h2.drive(settings_rx).await.unwrap();
let request = Request::get("https://example.com/a").body(()).unwrap();
tx.unbounded_send(client.send_request(request, true).unwrap().0)
tx.send(client.send_request(request, true).unwrap().0)
.unwrap();
let request = Request::get("https://example.com/b").body(()).unwrap();
tx.unbounded_send(client.send_request(request, true).unwrap().0)
tx.send(client.send_request(request, true).unwrap().0)
.unwrap();
tokio::spawn(async move {
@@ -760,8 +759,8 @@ async fn connection_notified_on_released_capacity() {
});
// Get the two requests
let a = rx.next().await.unwrap();
let b = rx.next().await.unwrap();
let a = rx.recv().await.unwrap();
let b = rx.recv().await.unwrap();
// Get the first response
let response = a.await.unwrap();
@@ -769,7 +768,7 @@ async fn connection_notified_on_released_capacity() {
let (_, mut a) = response.into_parts();
// Get the next chunk
let chunk = a.next().await.unwrap();
let chunk = a.data().await.unwrap();
assert_eq!(16_384, chunk.unwrap().len());
// Get the second response
@@ -778,7 +777,7 @@ async fn connection_notified_on_released_capacity() {
let (_, mut b) = response.into_parts();
// Get the next chunk
let chunk = b.next().await.unwrap();
let chunk = b.data().await.unwrap();
assert_eq!(16_384, chunk.unwrap().len());
// Wait a bit
@@ -944,7 +943,6 @@ async fn recv_no_init_window_then_receive_some_init_window() {
async fn settings_lowered_capacity_returns_capacity_to_connection() {
use futures::channel::oneshot;
use futures::future::{select, Either};
use std::time::Instant;
let _ = env_logger::try_init();
let (io, mut srv) = mock::new();
@@ -976,11 +974,7 @@ async fn settings_lowered_capacity_returns_capacity_to_connection() {
//
// A timeout is used here to avoid blocking forever if there is a
// failure
let result = select(
rx2,
tokio::timer::delay(Instant::now() + Duration::from_secs(5)),
)
.await;
let result = select(rx2, tokio::time::delay_for(Duration::from_secs(5))).await;
if let Either::Right((_, _)) = result {
panic!("Timed out");
}
@@ -1012,11 +1006,7 @@ async fn settings_lowered_capacity_returns_capacity_to_connection() {
});
// Wait for server handshake to complete.
let result = select(
rx1,
tokio::timer::delay(Instant::now() + Duration::from_secs(5)),
)
.await;
let result = select(rx1, tokio::time::delay_for(Duration::from_secs(5))).await;
if let Either::Right((_, _)) = result {
panic!("Timed out");
}
@@ -1113,7 +1103,7 @@ async fn increase_target_window_size_after_using_some() {
// drive an empty future to allow the WINDOW_UPDATE
// to go out while the response capacity is still in use.
conn.drive(yield_once()).await;
let _res = conn.drive(res.into_body().try_concat()).await;
let _res = conn.drive(util::concat(res.into_body())).await;
conn.await.expect("client");
};
@@ -1156,7 +1146,7 @@ async fn decrease_target_window_size() {
let mut body = res.into_parts().1;
let mut cap = body.flow_control().clone();
let bytes = conn.drive(body.try_concat()).await.expect("concat");
let bytes = conn.drive(util::concat(body)).await.expect("concat");
assert_eq!(bytes.len(), 65_535);
cap.release_capacity(bytes.len()).unwrap();
conn.await.expect("conn");
@@ -1568,7 +1558,7 @@ async fn data_padding() {
let resp = response.await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = resp.into_body();
let bytes = body.try_concat().await.unwrap();
let bytes = util::concat(body).await.unwrap();
assert_eq!(bytes.len(), 100);
};
join(async move { conn.await.expect("client") }, fut).await;

10
tests/h2-tests/tests/hammer.rs
View File

@@ -26,8 +26,8 @@ impl Server {
{
let mk_data = Arc::new(mk_data);
let rt = tokio::runtime::Runtime::new().unwrap();
let listener = rt
let mut rt = tokio::runtime::Runtime::new().unwrap();
let mut listener = rt
.block_on(TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0))))
.unwrap();
let addr = listener.local_addr().unwrap();
@@ -35,8 +35,8 @@ impl Server {
let reqs2 = reqs.clone();
let join = thread::spawn(move || {
let server = async move {
let mut incoming = listener.incoming();
while let Some(socket) = incoming.next().await {
loop {
let socket = listener.accept().await.map(|(s, _)| s);
let reqs = reqs2.clone();
let mk_data = mk_data.clone();
tokio::spawn(async move {
@@ -140,7 +140,7 @@ fn hammer_client_concurrency() {
})
});
let rt = tokio::runtime::Runtime::new().unwrap();
let mut rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(tcp);
println!("...done");
}

4
tests/h2-tests/tests/ping_pong.rs
View File

@@ -1,6 +1,6 @@
use futures::channel::oneshot;
use futures::future::join;
use futures::{StreamExt, TryStreamExt};
use futures::StreamExt;
use h2_support::assert_ping;
use h2_support::prelude::*;
@@ -84,7 +84,7 @@ async fn pong_has_highest_priority() {
assert_eq!(req.method(), "POST");
let body = req.into_parts().1;
let body = body.try_concat().await.expect("body");
let body = util::concat(body).await.expect("body");
assert_eq!(body.len(), data.len());
let res = Response::builder().status(200).body(()).unwrap();
stream.send_response(res, true).expect("response");

6
tests/h2-tests/tests/prioritization.rs
View File

@@ -8,7 +8,7 @@ use std::task::Context;
async fn single_stream_send_large_body() {
let _ = env_logger::try_init();
let payload = [0; 1024];
let payload = vec![0; 1024];
let mock = mock_io::Builder::new()
.handshake()
@@ -55,7 +55,7 @@ async fn single_stream_send_large_body() {
assert_eq!(stream.capacity(), payload.len());
// Send the data
stream.send_data(payload[..].into(), true).unwrap();
stream.send_data(payload.into(), true).unwrap();
// Get the response
let resp = h2.run(response).await.unwrap();
@@ -116,7 +116,7 @@ async fn multiple_streams_with_payload_greater_than_default_window() {
stream3.reserve_capacity(payload_clone.len());
assert_eq!(stream3.capacity(), 0);
stream1.send_data(payload_clone[..].into(), true).unwrap();
stream1.send_data(payload_clone.into(), true).unwrap();
// hold onto streams so they don't close
// stream1 doesn't close because response1 is used

2
tests/h2-tests/tests/push_promise.rs
View File

@@ -45,7 +45,7 @@ async fn recv_push_works() {
assert_eq!(request.into_parts().0.method, Method::GET);
let resp = response.await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let b = resp.into_body().try_concat().await.unwrap();
let b = util::concat(resp.into_body()).await.unwrap();
assert_eq!(b, "promised_data");
Ok(())
}

6
tests/h2-tests/tests/server.rs
View File

@@ -1,7 +1,7 @@
#![deny(warnings)]
use futures::future::{join, poll_fn};
use futures::{StreamExt, TryStreamExt};
use futures::StreamExt;
use h2_support::prelude::*;
use tokio::io::AsyncWriteExt;
@@ -526,7 +526,7 @@ async fn abrupt_shutdown() {
let (req, tx) = srv.next().await.unwrap().expect("server receives request");
let req_fut = async move {
let body = req.into_body().try_concat().await;
let body = util::concat(req.into_body()).await;
drop(tx);
let err = body.expect_err("request body should error");
assert_eq!(
@@ -608,7 +608,7 @@ async fn graceful_shutdown() {
let body = req.into_parts().1;
let body = async move {
let buf = body.try_concat().await.unwrap();
let buf = util::concat(body).await.unwrap();
assert!(buf.is_empty());
let rsp = http::Response::builder().status(200).body(()).unwrap();

17
tests/h2-tests/tests/stream_states.rs
View File

@@ -5,6 +5,7 @@ use futures::{FutureExt, StreamExt, TryStreamExt};
use h2_support::prelude::*;
use h2_support::util::yield_once;
use std::task::Poll;
use tokio::sync::oneshot;
#[tokio::test]
async fn send_recv_headers_only() {
@@ -80,7 +81,7 @@ async fn send_recv_data() {
assert_eq!(stream.capacity(), 5);
// Send the data
stream.send_data("hello", true).unwrap();
stream.send_data("hello".as_bytes(), true).unwrap();
// Get the response
let resp = h2.run(response).await.unwrap();
@@ -204,7 +205,7 @@ async fn errors_if_recv_frame_exceeds_max_frame_size() {
let resp = client.get("https://example.com/").await.expect("response");
assert_eq!(resp.status(), StatusCode::OK);
let body = resp.into_parts().1;
let res = body.try_concat().await;
let res = util::concat(body).await;
let err = res.unwrap_err();
assert_eq!(err.to_string(), "protocol error: frame with invalid size");
};
@@ -252,7 +253,7 @@ async fn configure_max_frame_size() {
let resp = client.get("https://example.com/").await.expect("response");
assert_eq!(resp.status(), StatusCode::OK);
let body = resp.into_parts().1;
let buf = body.try_concat().await.expect("body");
let buf = util::concat(body).await.expect("body");
assert_eq!(buf.len(), 16_385);
};
@@ -313,7 +314,7 @@ async fn recv_goaway_finishes_processed_streams() {
.expect("response");
assert_eq!(resp.status(), StatusCode::OK);
let body = resp.into_parts().1;
let buf = body.try_concat().await.expect("body");
let buf = util::concat(body).await.expect("body");
assert_eq!(buf.len(), 16_384);
};
@@ -702,7 +703,7 @@ async fn rst_while_closing() {
let (io, mut srv) = mock::new();
// Rendevous when we've queued a trailers frame
let (tx, rx) = crate::futures::channel::oneshot::channel();
let (tx, rx) = oneshot::channel();
let srv = async move {
let settings = srv.assert_client_handshake().await;
@@ -765,7 +766,7 @@ async fn rst_with_buffered_data() {
let (io, mut srv) = mock::new_with_write_capacity(73);
// Synchronize the client / server on response
let (tx, rx) = crate::futures::channel::oneshot::channel();
let (tx, rx) = oneshot::channel();
let srv = async move {
let settings = srv.assert_client_handshake().await;
@@ -817,7 +818,7 @@ async fn err_with_buffered_data() {
let (io, mut srv) = mock::new_with_write_capacity(73);
// Synchronize the client / server on response
let (tx, rx) = crate::futures::channel::oneshot::channel();
let (tx, rx) = oneshot::channel();
let srv = async move {
let settings = srv.assert_client_handshake().await;
@@ -872,7 +873,7 @@ async fn send_err_with_buffered_data() {
let (io, mut srv) = mock::new_with_write_capacity(73);
// Synchronize the client / server on response
let (tx, rx) = crate::futures::channel::oneshot::channel();
let (tx, rx) = oneshot::channel();
let srv = async move {
let settings = srv.assert_client_handshake().await;