rama 0.2.0

use futures::StreamExt;
use futures::future::{Either, ready};
use futures::stream::FuturesUnordered;
use h2_support::prelude::*;
use std::pin::Pin;
use std::task::Context;
use std::{io, panic};

#[tokio::test]
#[ignore]
async fn handshake() {
    h2_support::trace_init!();

    let mock = mock_io::Builder::new()
        .handshake()
        .write(SETTINGS_ACK)
        .build();

    let (_client, h2) = client::handshake(mock).await.unwrap();

    tracing::trace!("hands have been shook");

    // At this point, the connection should be closed
    h2.await.unwrap();
}

#[tokio::test]
#[ignore]
async fn client_other_thread() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://http2.akamai.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
    };

    let h2 = async move {
        let (mut client, h2) = client::handshake(io).await.unwrap();
        tokio::spawn(async move {
            let request = Request::builder()
                .uri("https://http2.akamai.com/")
                .body(())
                .unwrap();
            let _res = client
                .send_request(request, true)
                .unwrap()
                .0
                .await
                .expect("request");
        });
        h2.await.expect("h2");
    };
    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn recv_invalid_server_stream_id() {
    h2_support::trace_init!();

    let mock = mock_io::Builder::new()
        .handshake()
        // Write GET /
        .write(&[
            0, 0, 0x10, 1, 5, 0, 0, 0, 1, 0x82, 0x87, 0x41, 0x8B, 0x9D, 0x29, 0xAC, 0x4B, 0x8F,
            0xA8, 0xE9, 0x19, 0x97, 0x21, 0xE9, 0x84,
        ])
        .write(SETTINGS_ACK)
        // Read response
        .read(&[0, 0, 1, 1, 5, 0, 0, 0, 2, 137])
        // Write GO_AWAY
        .write(&[0, 0, 8, 7, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1])
        .build();

    let (mut client, h2) = client::handshake(mock).await.unwrap();

    // Send the request
    let request = Request::builder()
        .uri("https://http2.akamai.com/")
        .body(())
        .unwrap();

    tracing::info!("sending request");
    let (response, _) = client.send_request(request, true).unwrap();

    // The connection errors
    assert!(h2.await.is_err());

    // The stream errors
    assert!(response.await.is_err());
}

#[tokio::test]
#[ignore]
async fn request_stream_id_overflows() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let h2 = async move {
        let (mut client, mut h2) = client::Builder::new()
            .initial_stream_id(u32::MAX >> 1)
            .handshake::<_, Bytes>(io)
            .await
            .unwrap();
        let request = Request::builder()
            .method(Method::GET)
            .uri("https://example.com/")
            .body(())
            .unwrap();

        // first request is allowed
        let (response, _) = client.send_request(request, true).unwrap();
        let _x = h2.drive(response).await.unwrap();

        let request = Request::builder()
            .method(Method::GET)
            .uri("https://example.com/")
            .body(())
            .unwrap();
        // second cannot use the next stream id, it's over
        let poll_err = poll_fn(|cx| client.poll_ready(cx)).await.unwrap_err();
        assert_eq!(poll_err.to_string(), "user error: stream ID overflowed");

        let err = client.send_request(request, true).unwrap_err();
        assert_eq!(err.to_string(), "user error: stream ID overflowed");

        h2.await.unwrap();
    };

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(u32::MAX >> 1)
                .request("GET", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(u32::MAX >> 1).response(200).eos())
            .await;
        idle_ms(10).await;
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn client_builder_max_concurrent_streams() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let mut settings = frame::Settings::default();
    settings.set_max_concurrent_streams(Some(1));

    let srv = async move {
        let rcvd_settings = srv.assert_client_handshake().await;
        assert_frame_eq(settings, rcvd_settings);

        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
    };

    let mut builder = client::Builder::new();
    builder.max_concurrent_streams(1);

    let h2 = async move {
        let (mut client, mut h2) = builder.handshake::<_, Bytes>(io).await.unwrap();
        let request = Request::builder()
            .method(Method::GET)
            .uri("https://example.com/")
            .body(())
            .unwrap();
        let (response, _) = client.send_request(request, true).unwrap();
        h2.drive(response).await.unwrap();
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn request_over_max_concurrent_streams_errors() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv
            .assert_client_handshake_with_settings(
                frames::settings()
                    // super tiny server
                    .max_concurrent_streams(1),
            )
            .await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .request("POST", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
        srv.recv_frame(frames::headers(3).request("POST", "https://example.com/"))
            .await;
        srv.send_frame(frames::headers(3).response(200)).await;
        srv.recv_frame(frames::data(3, "hello").eos()).await;
        srv.send_frame(frames::data(3, "").eos()).await;
        srv.recv_frame(frames::headers(5).request("POST", "https://example.com/"))
            .await;
        srv.send_frame(frames::headers(5).response(200)).await;
        srv.recv_frame(frames::data(5, "hello").eos()).await;
        srv.send_frame(frames::data(5, "").eos()).await;
    };

    let h2 = async move {
        let (mut client, mut h2) = client::handshake(io).await.expect("handshake");
        // we send a simple req here just to drive the connection so we can
        // receive the server settings.
        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();
        // first request is allowed
        let (response, _) = client.send_request(request, true).unwrap();
        h2.drive(response).await.unwrap();

        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();

        // first request is allowed
        let (resp1, mut stream1) = client.send_request(request, false).unwrap();
        // as long as we let the connection internals tick
        client = h2.drive(client.ready()).await.unwrap();

        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();

        // second request is put into pending_open
        let (resp2, mut stream2) = client.send_request(request, false).unwrap();

        let request = Request::builder()
            .method(Method::GET)
            .uri("https://example.com/")
            .body(())
            .unwrap();

        let waker = futures::task::noop_waker();
        let mut cx = Context::from_waker(&waker);

        // third stream is over max concurrent
        assert!(!client.poll_ready(&mut cx).is_ready());

        let err = client.send_request(request, true).unwrap_err();
        assert_eq!(err.to_string(), "user error: rejected");

        stream1
            .send_data("hello".into(), true)
            .expect("req send_data");

        h2.drive(async move {
            resp1.await.expect("req");
            stream2
                .send_data("hello".into(), true)
                .expect("req2 send_data");
        })
        .await;
        join(async move { h2.await.unwrap() }, async move {
            resp2.await.unwrap()
        })
        .await;
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn recv_decrement_max_concurrent_streams_when_requests_queued() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .request("POST", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;

        srv.ping_pong([0; 8]).await;

        // limit this server later in life
        srv.send_frame(frames::settings().max_concurrent_streams(1))
            .await;
        srv.recv_frame(frames::settings_ack()).await;
        srv.recv_frame(
            frames::headers(3)
                .request("POST", "https://example.com/")
                .eos(),
        )
        .await;
        srv.ping_pong([1; 8]).await;
        srv.send_frame(frames::headers(3).response(200).eos()).await;

        srv.recv_frame(
            frames::headers(5)
                .request("POST", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(5).response(200).eos()).await;
    };

    let h2 = async move {
        let (mut client, mut h2) = client::handshake(io).await.expect("handshake");
        // we send a simple req here just to drive the connection so we can
        // receive the server settings.
        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();
        // first request is allowed
        let (response, _) = client.send_request(request, true).unwrap();
        h2.drive(response).await.unwrap();

        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();

        // first request is allowed
        let (resp1, _) = client.send_request(request, true).unwrap();

        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();

        // second request is put into pending_open
        let (resp2, _) = client.send_request(request, true).unwrap();

        h2.drive(async move {
            resp1.await.expect("req");
        })
        .await;
        join(async move { h2.await.unwrap() }, async move {
            resp2.await.unwrap()
        })
        .await;
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn send_request_poll_ready_when_connection_error() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv
            .assert_client_handshake_with_settings(
                frames::settings()
                    // super tiny server
                    .max_concurrent_streams(1),
            )
            .await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .request("POST", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
        srv.recv_frame(
            frames::headers(3)
                .request("POST", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(8).response(200).eos()).await;
    };

    let h2 = async move {
        let (mut client, mut h2) = client::handshake(io).await.expect("handshake");
        // we send a simple req here just to drive the connection so we can
        // receive the server settings.
        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();

        // first request is allowed
        let (response, _) = client.send_request(request, true).unwrap();
        h2.drive(response).await.unwrap();

        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();

        // first request is allowed
        let (resp1, _) = client.send_request(request, true).unwrap();
        // as long as we let the connection internals tick
        client = h2.drive(client.ready()).await.unwrap();

        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();

        // second request is put into pending_open
        let (resp2, _) = client.send_request(request, true).unwrap();

        // third stream is over max concurrent
        let until_ready = async move {
            poll_fn(move |cx| client.poll_ready(cx))
                .await
                .expect_err("client poll_ready");
        };

        // a FuturesUnordered is used on purpose!
        //
        // We don't want a join, since any of the other futures notifying
        // will make the until_ready future polled again, but we are
        // specifically testing that until_ready gets notified on its own.
        let mut unordered =
            futures::stream::FuturesUnordered::<Pin<Box<dyn Future<Output = ()>>>>::new();
        unordered.push(Box::pin(until_ready));
        unordered.push(Box::pin(async move {
            h2.await.expect_err("client conn");
        }));
        unordered.push(Box::pin(async move {
            resp1.await.expect_err("req1");
        }));
        unordered.push(Box::pin(async move {
            resp2.await.expect_err("req2");
        }));

        while unordered.next().await.is_some() {}
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn send_reset_notifies_recv_stream() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(frames::headers(1).request("POST", "https://example.com/"))
            .await;
        srv.send_frame(frames::headers(1).response(200)).await;
        srv.recv_frame(frames::reset(1).refused()).await;
        srv.recv_frame(frames::go_away(0)).await;
        srv.recv_eof().await;
    };

    let client = async move {
        let (mut client, mut conn) = client::handshake(io).await.expect("handshake");
        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();

        // first request is allowed
        let (resp1, mut tx) = client.send_request(request, false).unwrap();
        let res = conn.drive(resp1).await.unwrap();

        let tx = async move {
            tx.send_reset(h2::Reason::REFUSED_STREAM);
        };
        let rx = async {
            let mut body = res.into_body();
            let err = body.next().await.unwrap().expect_err("RecvBody");
            assert_eq!(
                err.to_string(),
                "stream error sent by user: refused stream before processing any application logic"
            );
        };

        // a FuturesUnordered is used on purpose!
        //
        // 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 unordered = FuturesUnordered::<Pin<Box<dyn Future<Output = ()>>>>::new();
        unordered.push(Box::pin(rx));
        unordered.push(Box::pin(tx));

        conn.drive(unordered.for_each(ready)).await;
        drop(client); // now let client gracefully goaway
        conn.await.expect("client");
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
async fn http_11_request_without_scheme_or_authority() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(frames::headers(1).request("GET", "/").scheme("http").eos())
            .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
    };

    let h2 = async move {
        let (mut client, mut h2) = client::handshake(io).await.expect("handshake");

        // HTTP_11 request with just :path is allowed
        let request = Request::builder()
            .method(Method::GET)
            .uri("/")
            .body(())
            .unwrap();

        let (response, _) = client.send_request(request, true).unwrap();
        h2.drive(response).await.unwrap();
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn http_2_request_without_scheme_or_authority() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
    };

    let h2 = async move {
        let (mut client, h2) = client::handshake(io).await.expect("handshake");

        // HTTP_2 with only a :path is illegal, so this request should
        // be rejected as a user error.
        let request = Request::builder()
            .version(Version::HTTP_2)
            .method(Method::GET)
            .uri("/")
            .body(())
            .unwrap();

        client
            .send_request(request, true)
            .expect_err("should be UserError");
        let _: () = h2.await.expect("h2");
        drop(client);
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn http_2_connect_request_omit_scheme_and_path_fields() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .pseudo(frame::Pseudo {
                    method: Method::CONNECT.into(),
                    authority: util::byte_str("tunnel.example.com:8443").into(),
                    ..Default::default()
                })
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
    };

    let h2 = async move {
        let (mut client, mut h2) = client::handshake(io).await.expect("handshake");

        // In HTTP_2 CONNECT request the ":scheme" and ":path" pseudo-header fields MUST be omitted.
        let request = Request::builder()
            .version(Version::HTTP_2)
            .method(Method::CONNECT)
            .uri("https://tunnel.example.com:8443/")
            .body(())
            .unwrap();

        let (response, _) = client.send_request(request, true).unwrap();
        h2.drive(response).await.unwrap();
    };

    join(srv, h2).await;
}

#[test]
#[ignore]
fn request_with_h1_version() {}

#[tokio::test]
#[ignore]
async fn request_with_connection_headers() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    // can't assert full handshake, since client never sends a request, and
    // thus never bothers to ack the settings...
    let srv = async move {
        srv.read_preface().await.unwrap();
        srv.recv_frame(frames::settings()).await;
        // goaway is required to make sure the connection closes because
        // of no active streams
        srv.recv_frame(frames::go_away(0)).await;
    };

    let headers = vec![
        ("connection", "foo"),
        ("keep-alive", "5"),
        ("proxy-connection", "bar"),
        ("transfer-encoding", "chunked"),
        ("upgrade", "HTTP/2"),
        ("te", "boom"),
    ];

    let client = async move {
        let (mut client, conn) = client::handshake(io).await.expect("handshake");

        for (name, val) in headers {
            let req = Request::builder()
                .uri("https://http2.akamai.com/")
                .header(name, val)
                .body(())
                .unwrap();
            let err = client.send_request(req, true).expect_err(name);
            assert_eq!(err.to_string(), "user error: malformed headers");
        }
        drop(client);
        conn.await.unwrap();
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
async fn connection_close_notifies_response_future() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();
    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://http2.akamai.com/")
                .eos(),
        )
        .await;
        // don't send any response, just close
    };

    let client = async move {
        let (mut client, conn) = client::handshake(io).await.expect("handshake");

        let request = Request::builder()
            .uri("https://http2.akamai.com/")
            .body(())
            .unwrap();

        let req = async move {
            let res = client
                .send_request(request, true)
                .expect("send_request1")
                .0
                .await;
            let err = res.expect_err("response");
            assert_eq!(err.to_string(), "stream closed because of a broken pipe");
        };
        join(async move { conn.await.expect("conn") }, req).await;
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
async fn connection_close_notifies_client_poll_ready() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://http2.akamai.com/")
                .eos(),
        )
        .await;
    };

    let client = async move {
        let (mut client, mut conn) = client::handshake(io).await.expect("handshake");

        let request = Request::builder()
            .uri("https://http2.akamai.com/")
            .body(())
            .unwrap();

        let req = async {
            let res = client
                .send_request(request, true)
                .expect("send_request1")
                .0
                .await;
            let err = res.expect_err("response");
            assert_eq!(err.to_string(), "stream closed because of a broken pipe");
        };

        conn.drive(req).await;

        let err = poll_fn(move |cx| client.poll_ready(cx))
            .await
            .expect_err("poll_ready");
        assert_eq!(
            err.to_string(),
            "connection closed because of a broken pipe"
        );
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
async fn sending_request_on_closed_connection() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://http2.akamai.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
        // a bad frame!
        srv.send_frame(frames::headers(0).response(200).eos()).await;
    };

    let h2 = async move {
        let (mut client, h2) = client::handshake(io).await.expect("handshake");

        let request = Request::builder()
            .uri("https://http2.akamai.com/")
            .body(())
            .unwrap();

        // first request works
        let req = Box::pin(async {
            client
                .send_request(request, true)
                .expect("send_request1")
                .0
                .await
                .expect("response1");
        });

        // after finish request1, there should be a conn error
        let h2 = Box::pin(async move {
            h2.await.expect_err("h2 error");
        });

        match select(h2, req).await {
            Either::Left((_, req)) => req.await,
            Either::Right((_, _h2)) => unreachable!("Shouldn't happen"), // TODO: Is this correct?
        };

        let poll_err = poll_fn(|cx| client.poll_ready(cx)).await.unwrap_err();
        let msg = "connection error detected: unspecific protocol error detected";
        assert_eq!(poll_err.to_string(), msg);

        let request = Request::builder()
            .uri("https://http2.akamai.com/")
            .body(())
            .unwrap();
        let send_err = client.send_request(request, true).unwrap_err();
        assert_eq!(send_err.to_string(), msg);
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn recv_too_big_headers() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_frame_eq(settings, frames::settings().max_header_list_size(10));
        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://http2.akamai.com/")
                .eos(),
        )
        .await;
        srv.recv_frame(
            frames::headers(3)
                .request("GET", "https://http2.akamai.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
        srv.send_frame(frames::headers(3).response(200)).await;
        // no reset for 1, since it's closed anyway
        // but reset for 3, since server hasn't closed stream
        srv.recv_frame(frames::reset(3).protocol_error()).await;
        idle_ms(10).await;
    };

    let client = async move {
        let (mut client, mut conn) = client::Builder::new()
            .max_header_list_size(10)
            .handshake::<_, Bytes>(io)
            .await
            .expect("handshake");

        let request = Request::builder()
            .uri("https://http2.akamai.com/")
            .body(())
            .unwrap();

        let req1 = client.send_request(request, true);
        // Spawn tasks to ensure that the error wakes up tasks that are blocked
        // waiting for a response.
        let req1 = async move {
            let err = req1.expect("send_request").0.await.expect_err("response1");
            assert_eq!(err.reason(), Some(Reason::PROTOCOL_ERROR));
        };

        let request = Request::builder()
            .uri("https://http2.akamai.com/")
            .body(())
            .unwrap();

        let req2 = client.send_request(request, true);
        let req2 = async move {
            let err = req2.expect("send_request").0.await.expect_err("response2");
            assert_eq!(err.reason(), Some(Reason::PROTOCOL_ERROR));
        };

        conn.drive(join(req1, req2)).await;
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
async fn pending_send_request_gets_reset_by_peer_properly() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    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 {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(frames::headers(1).request("GET", "https://http2.akamai.com/"))
            .await;
        // Note that we can only send up to ~4 frames of data by default
        srv.recv_frame(frames::data(1, &payload[0..max_frame_size]))
            .await;
        srv.recv_frame(frames::data(
            1,
            &payload[max_frame_size..(max_frame_size * 2)],
        ))
        .await;
        srv.recv_frame(frames::data(
            1,
            &payload[(max_frame_size * 2)..(max_frame_size * 3)],
        ))
        .await;
        srv.recv_frame(frames::data(
            1,
            &payload[(max_frame_size * 3)..(max_frame_size * 4 - 1)],
        ))
        .await;

        idle_ms(100).await;

        srv.send_frame(frames::reset(1).refused()).await;
        // Because all active requests are finished, connection should shutdown
        // and send a GO_AWAY frame. If the reset stream is bugged (and doesn't
        // count towards concurrency limit), then connection will not send
        // a GO_AWAY and this test will fail.
        srv.recv_frame(frames::go_away(0)).await;
        drop(srv);
    };

    let client = async {
        let (mut client, mut conn) = client::Builder::new()
            .handshake::<_, Bytes>(io)
            .await
            .expect("handshake");

        let request = Request::builder()
            .uri("https://http2.akamai.com/")
            .body(())
            .unwrap();

        let (response, mut stream) = client.send_request(request, false).expect("send_request");

        let response = async move {
            let err = response.await.expect_err("response");
            assert_eq!(err.reason(), Some(Reason::REFUSED_STREAM));
        };

        // Send the data
        stream.send_data(payload.clone(), true).unwrap();
        conn.drive(response).await;
        drop(client);
        drop(stream);
        conn.await.expect("client");
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
async fn request_without_path() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);

        srv.recv_frame(
            frames::headers(1)
                .request("GET", "http://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
    };

    let client = async move {
        let (mut client, mut conn) = client::handshake(io).await.expect("handshake");
        // Note the lack of trailing slash.
        let request = Request::get("http://example.com").body(()).unwrap();

        let (response, _) = client.send_request(request, true).unwrap();

        conn.drive(response).await.unwrap();
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
async fn request_options_with_star() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    // Note the lack of trailing slash.
    let uri = uri::Uri::from_parts({
        let mut parts = uri::Parts::default();
        parts.scheme = Some(uri::Scheme::HTTP);
        parts.authority = Some(uri::Authority::from_static("example.com"));
        parts.path_and_query = Some(uri::PathAndQuery::from_static("*"));
        parts
    })
    .unwrap();

    let uri_clone = uri.clone();
    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(frames::headers(1).request("OPTIONS", uri_clone).eos())
            .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
    };

    let client = async move {
        let (mut client, mut conn) = client::handshake(io).await.expect("handshake");
        let request = Request::builder()
            .method(Method::OPTIONS)
            .uri(uri)
            .body(())
            .unwrap();

        let (response, _) = client.send_request(request, true).unwrap();

        conn.drive(response).await.unwrap();
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
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 tokio::sync::oneshot;

    h2_support::trace_init!();

    let (io, mut srv) = mock::new();
    let (done_tx, done_rx) = oneshot::channel();
    let (tx, rx) = oneshot::channel();

    let mut settings = frame::Settings::default();
    settings.set_max_concurrent_streams(Some(1));

    let srv = async move {
        let settings = srv.assert_client_handshake_with_settings(settings).await;
        // This is the ACK
        assert_default_settings!(settings);
        tx.send(()).unwrap();
        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://www.example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
        srv.recv_frame(
            frames::headers(3)
                .request("GET", "https://www.example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(3).response(200).eos()).await;
        srv.recv_frame(
            frames::headers(5)
                .request("GET", "https://www.example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(5).response(200).eos()).await;
        // Don't close the connection until the client is done doing its
        // checks.
        done_rx.await.unwrap();
    };

    let client = async move {
        let (mut client, conn) = client::handshake(io).await.expect("handshake");
        tokio::spawn(async move {
            rx.await.unwrap();

            let mut responses = vec![];

            for _ in 0..3usize {
                // Wait for capacity. If the client is **not** notified,
                // this hangs.
                poll_fn(|cx| client.poll_ready(cx)).await.unwrap();

                let request = Request::builder()
                    .uri("https://www.example.com/")
                    .body(())
                    .unwrap();

                let response = client.send_request(request, true).unwrap().0;

                responses.push(response);
            }

            for response in responses {
                let response = response.await.unwrap();
                assert_eq!(response.status(), StatusCode::OK);
            }

            poll_fn(|cx| client.poll_ready(cx)).await.unwrap();

            done_tx.send(()).unwrap();
        });

        conn.await.expect("h2");
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
async fn send_stream_poll_reset() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(frames::headers(1).request("POST", "https://example.com/"))
            .await;
        srv.send_frame(frames::reset(1).refused()).await;
    };

    let client = async move {
        let (mut client, mut conn) = client::Builder::new()
            .handshake::<_, Bytes>(io)
            .await
            .expect("handshake");
        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();

        let (_response, mut tx) = client.send_request(request, false).unwrap();
        let reason = conn
            .drive(poll_fn(move |cx| tx.poll_reset(cx)))
            .await
            .unwrap();
        assert_eq!(reason, Reason::REFUSED_STREAM);
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
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;
    h2_support::trace_init!();

    let (io, mut srv) = mock::new();
    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));

    let srv = async move {
        let settings = srv.assert_client_handshake_with_settings(settings).await;
        // This is the ACK
        assert_default_settings!(settings);
        init_tx.send(()).unwrap();
        srv.recv_frame(frames::headers(1).request("GET", "https://www.example.com/"))
            .await;
        srv.recv_frame(
            frames::headers(3)
                .request("GET", "https://www.example.com/")
                .eos(),
        )
        .await;
        trigger_go_away_rx.await.unwrap();
        srv.send_frame(frames::go_away(3)).await;
        sent_go_away_tx.send(()).unwrap();
        drop_rx.await.unwrap();
        srv.send_frame(frames::headers(3).response(200).eos()).await;
        srv.recv_frame(frames::data(1, vec![]).eos()).await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
    };

    fn request() -> Request<()> {
        Request::builder()
            .uri("https://www.example.com/")
            .body(())
            .unwrap()
    }

    let client = async move {
        let (mut client, conn) = client::Builder::new()
            .max_concurrent_reset_streams(0)
            .handshake::<_, Bytes>(io)
            .await
            .expect("handshake");
        let f = async move {
            init_rx.await.expect("init_rx");
            // Fill up the concurrent stream limit.
            poll_fn(|cx| client.poll_ready(cx)).await.unwrap();
            let mut response1 = client.send_request(request(), false).unwrap();
            poll_fn(|cx| client.poll_ready(cx)).await.unwrap();
            let response2 = client.send_request(request(), true).unwrap();
            poll_fn(|cx| client.poll_ready(cx)).await.unwrap();
            let response3 = client.send_request(request(), true).unwrap();

            // Trigger a GOAWAY frame to invalidate our third request.
            trigger_go_away_tx.send(()).unwrap();
            sent_go_away_rx.await.expect("sent_go_away_rx");
            // Now drop all the references to that stream.
            drop(response3);
            drop(client);
            drop_tx.send(()).unwrap();

            // Complete the second request, freeing up a stream.
            response2.0.await.expect("resp2");
            response1.1.send_data(Default::default(), true).unwrap();
            response1.0.await.expect("resp1")
        };

        join(
            async move {
                conn.await.expect("h2");
            },
            f,
        )
        .await;
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
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;
    h2_support::trace_init!();

    let (io, mut srv) = mock::new();
    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;
        assert_default_settings!(settings);

        srv.recv_frame(frames::headers(1).request("GET", "http://example.com/"))
            .await;
        srv.recv_frame(frames::headers(3).request("GET", "http://example.com/"))
            .await;
        srv.recv_frame(frames::headers(5).request("GET", "http://example.com/"))
            .await;
        drop_tx.send(()).unwrap();
        queued_rx.await.unwrap();
        srv.send_bytes(&[
            // 2 byte frame
            0, 0, 2, // type: HEADERS
            1, // flags: END_STREAM | END_HEADERS
            5, // stream identifier: 3
            0, 0, 0, 3, // data - invalid (pseudo not at end of block)
            144,
            135, // Per the spec, this frame should cause a stream error of type
                 // PROTOCOL_ERROR.
        ])
        .await;
    };

    fn request() -> Request<()> {
        Request::builder()
            .uri("http://example.com/")
            .body(())
            .unwrap()
    }

    let client = async move {
        let (mut client, conn) = client::Builder::new()
            .handshake::<_, Bytes>(io)
            .await
            .expect("handshake");

        let (_req1, mut send1) = client.send_request(request(), false).unwrap();
        // Use up most of the connection window.
        send1.send_data(vec![0; 65534].into(), true).unwrap();
        let (req2, mut send2) = client.send_request(request(), false).unwrap();
        let (req3, mut send3) = client.send_request(request(), false).unwrap();

        let f = async move {
            drop_rx.await.unwrap();
            // Use up the remainder of the connection window.
            send2.send_data(vec![0; 2].into(), true).unwrap();
            // Queue up for more connection window.
            send3.send_data(vec![0; 1].into(), true).unwrap();
            queued_tx.send(()).unwrap();
            drop((req2, req3));
        };

        join(async move { conn.await.expect("h2") }, f).await;
    };

    join(srv, client).await;
}

#[tokio::test]
#[ignore]
async fn allow_empty_data_for_head() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .request("HEAD", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(
            frames::headers(1)
                .response(200)
                .field("content-length", 100),
        )
        .await;
        srv.send_frame(frames::data(1, "").eos()).await;
    };

    let h2 = async move {
        let (mut client, h2) = client::Builder::new()
            .handshake::<_, Bytes>(io)
            .await
            .unwrap();
        tokio::spawn(async {
            h2.await.expect("connection failed");
        });
        let request = Request::builder()
            .method(Method::HEAD)
            .uri("https://example.com/")
            .body(())
            .unwrap();
        let (response, _) = client.send_request(request, true).unwrap();
        let (_, mut body) = response.await.unwrap().into_parts();
        assert_eq!(body.data().await.unwrap().unwrap(), "");
    };

    join(srv, h2).await;
}

#[tokio::test]
async fn reject_none_zero_content_length_header_with_end_stream() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(
            frames::headers(1)
                .response(200)
                .field("content-length", 100)
                .eos(),
        )
        .await;
    };

    let h2 = async move {
        let (mut client, h2) = client::Builder::new()
            .handshake::<_, Bytes>(io)
            .await
            .unwrap();
        tokio::spawn(async {
            h2.await.expect("connection failed");
        });
        let request = Request::builder()
            .method(Method::GET)
            .uri("https://example.com/")
            .body(())
            .unwrap();
        let (response, _) = client.send_request(request, true).unwrap();
        let _ = response.await.unwrap_err();
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn early_hints() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(103)).await;
        srv.send_frame(frames::headers(1).response(200).field("content-length", 2))
            .await;
        srv.send_frame(frames::data(1, "ok").eos()).await;
    };

    let h2 = async move {
        let (mut client, h2) = client::Builder::new()
            .handshake::<_, Bytes>(io)
            .await
            .unwrap();
        tokio::spawn(async {
            h2.await.expect("connection failed");
        });
        let request = Request::builder()
            .method(Method::GET)
            .uri("https://example.com/")
            .body(())
            .unwrap();
        let (response, _) = client.send_request(request, true).unwrap();
        let (ha, mut body) = response.await.unwrap().into_parts();
        eprintln!("{:?}", ha);
        assert_eq!(body.data().await.unwrap().unwrap(), "ok");
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn informational_while_local_streaming() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(frames::headers(1).request("POST", "https://example.com/"))
            .await;
        srv.send_frame(frames::headers(1).response(103)).await;
        srv.send_frame(frames::headers(1).response(200).field("content-length", 2))
            .await;
        srv.recv_frame(frames::data(1, "hello").eos()).await;
        srv.send_frame(frames::data(1, "ok").eos()).await;
    };

    let h2 = async move {
        let (mut client, h2) = client::Builder::new()
            .handshake::<_, Bytes>(io)
            .await
            .unwrap();
        tokio::spawn(async {
            h2.await.expect("connection failed");
        });
        let request = Request::builder()
            .method(Method::POST)
            .uri("https://example.com/")
            .body(())
            .unwrap();
        // don't EOS stream yet..
        let (response, mut body_tx) = client.send_request(request, false).unwrap();
        // eventual response is 200, not 103
        let resp = response.await.expect("response");
        // assert_eq!(resp.status(), 200);
        // now we can end the stream
        body_tx.send_data("hello".into(), true).expect("send_data");
        let mut body = resp.into_body();
        assert_eq!(body.data().await.unwrap().unwrap(), "ok");
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn extended_connect_protocol_disabled_by_default() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);

        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
    };

    let h2 = async move {
        let (mut client, mut h2) = client::handshake(io).await.unwrap();

        // we send a simple req here just to drive the connection so we can
        // receive the server settings.
        let request = Request::get("https://example.com/").body(()).unwrap();
        // first request is allowed
        let (response, _) = client.send_request(request, true).unwrap();
        h2.drive(response).await.unwrap();

        assert!(!client.is_extended_connect_protocol_enabled());
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn extended_connect_protocol_enabled_during_handshake() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv
            .assert_client_handshake_with_settings(frames::settings().enable_connect_protocol(1))
            .await;
        assert_default_settings!(settings);

        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
    };

    let h2 = async move {
        let (mut client, mut h2) = client::handshake(io).await.unwrap();

        // we send a simple req here just to drive the connection so we can
        // receive the server settings.
        let request = Request::get("https://example.com/").body(()).unwrap();
        let (response, _) = client.send_request(request, true).unwrap();
        h2.drive(response).await.unwrap();

        assert!(client.is_extended_connect_protocol_enabled());
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn invalid_connect_protocol_enabled_setting() {
    h2_support::trace_init!();

    let (io, mut srv) = mock::new();

    let srv = async move {
        // Send a settings frame
        srv.send(frames::settings().enable_connect_protocol(2).into())
            .await
            .unwrap();
        srv.read_preface().await.unwrap();

        let settings = assert_settings!(srv.next().await.expect("unexpected EOF").unwrap());
        assert_default_settings!(settings);

        // Send the ACK
        let ack = frame::Settings::ack();

        // TODO: Don't unwrap?
        srv.send(ack.into()).await.unwrap();

        let frame = srv.next().await.unwrap().unwrap();
        let go_away = assert_go_away!(frame);
        assert_eq!(go_away.reason(), Reason::PROTOCOL_ERROR);
    };

    let h2 = async move {
        let (mut client, mut h2) = client::handshake(io).await.unwrap();

        // we send a simple req here just to drive the connection so we can
        // receive the server settings.
        let request = Request::get("https://example.com/").body(()).unwrap();
        let (response, _) = client.send_request(request, true).unwrap();

        let error = h2.drive(response).await.unwrap_err();
        assert_eq!(error.reason(), Some(Reason::PROTOCOL_ERROR));
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn extended_connect_request() {
    h2_support::trace_init!();

    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv
            .assert_client_handshake_with_settings(frames::settings().enable_connect_protocol(1))
            .await;
        assert_default_settings!(settings);

        srv.recv_frame(
            frames::headers(1)
                .pseudo(frame::Pseudo {
                    method: Method::CONNECT.into(),
                    scheme: util::byte_str("http").into(),
                    authority: util::byte_str("bread").into(),
                    path: util::byte_str("/baguette").into(),
                    protocol: Protocol::from_static("the-bread-protocol").into(),
                    ..Default::default()
                })
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
    };

    let h2 = async move {
        let (mut client, mut h2) = client::handshake(io).await.unwrap();

        let request = Request::connect("http://bread/baguette")
            .extension(Protocol::from("the-bread-protocol"))
            .body(())
            .unwrap();
        let (response, _) = client.send_request(request, true).unwrap();
        h2.drive(response).await.unwrap();
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn rogue_server_odd_headers() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.send_frame(frames::headers(1)).await;
        srv.recv_frame(frames::go_away(0).protocol_error()).await;
    };

    let h2 = async move {
        let (_client, h2) = client::handshake(io).await.unwrap();

        let err = h2.await.unwrap_err();
        assert!(err.is_go_away());
        assert_eq!(err.reason(), Some(Reason::PROTOCOL_ERROR));
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn rogue_server_even_headers() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.send_frame(frames::headers(2)).await;
        srv.recv_frame(frames::go_away(0).protocol_error()).await;
    };

    let h2 = async move {
        let (_client, h2) = client::handshake(io).await.unwrap();

        let err = h2.await.unwrap_err();
        assert!(err.is_go_away());
        assert_eq!(err.reason(), Some(Reason::PROTOCOL_ERROR));
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn rogue_server_reused_headers() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);

        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://camembert.fromage")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
        srv.send_frame(frames::headers(1)).await;
        srv.recv_frame(frames::reset(1).stream_closed()).await;
    };

    let h2 = async move {
        let (mut client, mut h2) = client::handshake(io).await.unwrap();

        h2.drive(async {
            let request = Request::builder()
                .method(Method::GET)
                .uri("https://camembert.fromage")
                .body(())
                .unwrap();
            let _res = client.send_request(request, true).unwrap().0.await.unwrap();
        })
        .await;

        h2.await.unwrap();
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn client_builder_header_table_size() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();
    let mut settings = frame::Settings::default();

    settings.set_header_table_size(Some(10000));

    let srv = async move {
        let recv_settings = srv.assert_client_handshake().await;
        assert_frame_eq(recv_settings, settings);

        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://example.com/")
                .eos(),
        )
        .await;
        srv.send_frame(frames::headers(1).response(200).eos()).await;
    };

    let mut builder = client::Builder::new();
    builder.header_table_size(10000);

    let h2 = async move {
        let (mut client, mut h2) = builder.handshake::<_, Bytes>(io).await.unwrap();
        let request = Request::get("https://example.com/").body(()).unwrap();
        let (response, _) = client.send_request(request, true).unwrap();
        h2.drive(response).await.unwrap();
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn configured_max_concurrent_send_streams_and_update_it_based_on_empty_settings_frame() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        // Send empty SETTINGS frame (no MAX_CONCURRENT_STREAMS is provided)
        srv.send_frame(frames::settings()).await;
    };

    let h2 = async move {
        let (_client, h2) = client::Builder::new()
            // Configure the initial value to 2024
            .initial_max_send_streams(2024)
            .handshake::<_, bytes::Bytes>(io)
            .await
            .unwrap();
        let mut h2 = std::pin::pin!(h2);
        // It should be pre-configured value before it receives the initial
        // SETTINGS frame from the server
        assert_eq!(h2.max_concurrent_send_streams(), 2024);
        h2.as_mut().await.unwrap();
        // If the server's initial SETTINGS frame does not include
        // MAX_CONCURRENT_STREAMS, this should be updated to usize::MAX.
        assert_eq!(h2.max_concurrent_send_streams(), usize::MAX);
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn configured_max_concurrent_send_streams_and_update_it_based_on_non_empty_settings_frame() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        // Send SETTINGS frame with MAX_CONCURRENT_STREAMS set to 42
        srv.send_frame(frames::settings().max_concurrent_streams(42))
            .await;
    };

    let h2 = async move {
        let (_client, h2) = client::Builder::new()
            // Configure the initial value to 2024
            .initial_max_send_streams(2024)
            .handshake::<_, bytes::Bytes>(io)
            .await
            .unwrap();
        let mut h2 = std::pin::pin!(h2);
        // It should be pre-configured value before it receives the initial
        // SETTINGS frame from the server
        assert_eq!(h2.max_concurrent_send_streams(), 2024);
        h2.as_mut().await.unwrap();
        // Now the client has received the initial SETTINGS frame from the
        // server, which should update the value accordingly
        assert_eq!(h2.max_concurrent_send_streams(), 42);
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn receive_settings_frame_twice_with_second_one_empty() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        // Send the initial SETTINGS frame with MAX_CONCURRENT_STREAMS set to 42
        srv.send_frame(frames::settings().max_concurrent_streams(42))
            .await;

        // Handle the client's connection preface
        srv.read_preface().await.unwrap();
        match srv.next().await {
            Some(frame) => match frame.unwrap() {
                h2::frame::Frame::Settings(_) => {
                    let ack = frame::Settings::ack();
                    srv.send(ack.into()).await.unwrap();
                }
                frame => {
                    panic!("unexpected frame: {:?}", frame);
                }
            },
            None => {
                panic!("unexpected EOF");
            }
        }

        // Should receive the ack for the server's initial SETTINGS frame
        let frame = assert_settings!(srv.next().await.unwrap().unwrap());
        assert!(frame.is_ack());

        // Send another SETTINGS frame with no MAX_CONCURRENT_STREAMS
        // This should not update the max_concurrent_send_streams value that
        // the client manages.
        srv.send_frame(frames::settings()).await;
    };

    let h2 = async move {
        let (_client, h2) = client::handshake(io).await.unwrap();
        let mut h2 = std::pin::pin!(h2);
        assert_eq!(h2.max_concurrent_send_streams(), usize::MAX);
        h2.as_mut().await.unwrap();
        // Even though the second SETTINGS frame contained no value for
        // MAX_CONCURRENT_STREAMS, update to usize::MAX should not happen
        assert_eq!(h2.max_concurrent_send_streams(), 42);
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn receive_settings_frame_twice_with_second_one_non_empty() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        // Send the initial SETTINGS frame with MAX_CONCURRENT_STREAMS set to 42
        srv.send_frame(frames::settings().max_concurrent_streams(42))
            .await;

        // Handle the client's connection preface
        srv.read_preface().await.unwrap();
        match srv.next().await {
            Some(frame) => match frame.unwrap() {
                h2::frame::Frame::Settings(_) => {
                    let ack = frame::Settings::ack();
                    srv.send(ack.into()).await.unwrap();
                }
                frame => {
                    panic!("unexpected frame: {:?}", frame);
                }
            },
            None => {
                panic!("unexpected EOF");
            }
        }

        // Should receive the ack for the server's initial SETTINGS frame
        let frame = assert_settings!(srv.next().await.unwrap().unwrap());
        assert!(frame.is_ack());

        // Send another SETTINGS frame with no MAX_CONCURRENT_STREAMS
        // This should not update the max_concurrent_send_streams value that
        // the client manages.
        srv.send_frame(frames::settings().max_concurrent_streams(2024))
            .await;
    };

    let h2 = async move {
        let (_client, h2) = client::handshake(io).await.unwrap();
        let mut h2 = std::pin::pin!(h2);
        assert_eq!(h2.max_concurrent_send_streams(), usize::MAX);
        h2.as_mut().await.unwrap();
        // The most-recently advertised value should be used
        assert_eq!(h2.max_concurrent_send_streams(), 2024);
    };

    join(srv, h2).await;
}

#[tokio::test]
#[ignore]
async fn server_drop_connection_unexpectedly_return_unexpected_eof_err() {
    h2_support::trace_init!();
    let (io, mut srv) = mock::new();

    let srv = async move {
        let settings = srv.assert_client_handshake().await;
        assert_default_settings!(settings);
        srv.recv_frame(
            frames::headers(1)
                .request("GET", "https://http2.akamai.com/")
                .eos(),
        )
        .await;
        srv.close_without_notify();
    };

    let h2 = async move {
        let (mut client, h2) = client::handshake(io).await.unwrap();
        tokio::spawn(async move {
            let request = Request::builder()
                .uri("https://http2.akamai.com/")
                .body(())
                .unwrap();
            let _res = client
                .send_request(request, true)
                .unwrap()
                .0
                .await
                .expect("request");
        });
        let err = h2.await.expect_err("should receive UnexpectedEof");
        assert_eq!(
            err.get_io().expect("should be UnexpectedEof").kind(),
            io::ErrorKind::UnexpectedEof,
        );
    };
    join(srv, h2).await;
}

const SETTINGS: &[u8] = &[0, 0, 0, 4, 0, 0, 0, 0, 0];
const SETTINGS_ACK: &[u8] = &[0, 0, 0, 4, 1, 0, 0, 0, 0];

trait MockH2 {
    fn handshake(&mut self) -> &mut Self;
}

impl MockH2 for mock_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
            .write(SETTINGS)
            .read(SETTINGS)
            .read(SETTINGS_ACK)
    }
}