ETKNeil
commented
1 year ago I got the same error with h2:
Cargo.toml
[package]
name = "temp"
version = "0.1.0"
edition = "2021"
[dependencies]
tokio = { version = "1.21.2", features = ["rt-multi-thread", "macros", "time"] }
hyper = { version = "1.0.0-rc.1", features = ["client", "http2", "server", "http1"] }
bytes = "1.2.1"
http-body-util = "0.1.0-rc.1"
flume = "0.10.14"
uuid = { version = "1.2.2", features = ["v4"] }
h2 = { version = "0.3.15" }
env_logger = "0.9.3"H2
use bytes::Bytes;
use hyper::{http, HeaderMap, Request};
use std::io::Error;
use std::fmt::Debug;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tokio::time::Instant;
use uuid::Uuid;
#[derive(Clone, Debug)]
pub struct LocalExec {
handle: tokio::runtime::Handle,
}
impl LocalExec {
pub fn new(handle: tokio::runtime::Handle) -> Self {
Self { handle }
}
}
impl<F> hyper::rt::Executor<F> for LocalExec
where
F: Future + 'static + Send,
<F as Future>::Output: Debug,
<F as Future>::Output: Send,
{
fn execute(&self, fut: F) {
let uuid = Uuid::new_v4();
println!("Running task {}", uuid);
self.handle.spawn(async move {
let r = fut.await;
println!("Got output for task {} : {:?}", uuid, r)
});
}
}
pub(crate) struct TokioSleep {
pub(crate) inner: Pin<Box<tokio::time::Sleep>>,
}
impl Default for TokioSleep {
fn default() -> Self {
Self {
inner: Box::pin(tokio::time::sleep(Duration::default())),
}
}
}
impl Future for TokioSleep {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.inner.as_mut().poll(cx)
}
}
impl hyper::rt::Sleep for TokioSleep {}
pub struct IO {
name: String,
rx: flume::Receiver<Vec<u8>>,
tx: flume::Sender<Vec<u8>>,
sleep: TokioSleep,
}
impl IO {
pub fn new(name: String, rx: flume::Receiver<Vec<u8>>, tx: flume::Sender<Vec<u8>>) -> Self {
let sleep = TokioSleep::default();
Self {
name,
rx,
tx,
sleep,
}
}
}
impl AsyncRead for IO {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
let this = self.get_mut();
println!("Polling read from {}", this.name);
let mut sleep = Pin::new(&mut this.sleep);
match sleep.as_mut().poll(cx) {
Poll::Ready(_) => {
sleep
.inner
.as_mut()
.reset(Instant::now() + Duration::from_millis(25));
match this.rx.try_recv() {
Ok(msg) => {
buf.put_slice(msg.as_slice());
println!("Reading {}", String::from_utf8_lossy(buf.filled()));
Poll::Ready(Ok(()))
}
Err(_) => Poll::Pending,
}
}
Poll::Pending => Poll::Pending,
}
}
}
impl AsyncWrite for IO {
fn poll_write(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<Result<usize, Error>> {
let len = buf.len();
println!(
"Writing from {} {}",
&self.name,
String::from_utf8_lossy(buf)
);
match self.tx.send(buf.to_vec()) {
Err(err) => {
eprintln!("Could not send, the underlying channel might be dropped (disconnected:{}), got error {}", self.tx.is_disconnected(), err);
}
_ => {}
}
Poll::Ready(Ok(len))
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
println!("Flushing from {}", &self.name);
Poll::Ready(Ok(()))
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
println!("Shutdown from {}", &self.name);
Poll::Ready(Ok(()))
}
}
async fn handle_request(
mut request: Request<h2::RecvStream>,
mut respond: h2::server::SendResponse<Bytes>,
) -> Result<(), h2::Error> {
println!("GOT request: {:?}", request);
let body = request.body_mut();
while let Some(data) = body.data().await {
let data = data?;
println!("<<<< recv {:?}", data);
let _ = body.flow_control().release_capacity(data.len());
}
let response = http::Response::new(());
let mut send = respond.send_response(response, false)?;
println!(">>>> send");
send.send_data(Bytes::from_static(b"hello "), false)?;
send.send_data(Bytes::from_static(b"world\n"), true)?;
Ok(())
}
#[tokio::main(worker_threads = 2)]
pub async fn main() -> Result<(), h2::Error> {
let (server_tx, client_rx) = flume::unbounded();
let (client_tx, server_rx) = flume::unbounded();
let _ = env_logger::try_init();
let io = IO::new(String::from("Server"), server_rx, server_tx);
tokio::task::spawn(async move {
let mut connection = h2::server::handshake(io).await.unwrap();
while let Some(result) = connection.accept().await {
let (request, respond) = result.unwrap();
tokio::spawn(async move {
if let Err(e) = handle_request(request, respond).await {
println!("error while handling request: {:?}", e);
}
});
}
});
let io = IO::new(String::from("Client"), client_rx, client_tx);
let (mut client, h2) = h2::client::handshake(io).await.unwrap();
println!("sending request");
let request = Request::builder().uri("test").body(()).unwrap();
let mut trailers = HeaderMap::new();
trailers.insert("zomg", "hello".parse().unwrap());
let (response, mut stream) = client.send_request(request, false).unwrap();
// send trailers
stream.send_trailers(trailers).unwrap();
// Spawn a task to run the conn...
tokio::spawn(async move {
if let Err(e) = h2.await {
println!("GOT ERR={:?}", e);
}
});
let response = response.await?;
println!("GOT RESPONSE: {:?}", response);
// Get the body
let mut body = response.into_body();
while let Some(chunk) = body.data().await {
println!("GOT CHUNK = {:?}", chunk?);
}
if let Some(trailers) = body.trailers().await? {
println!("GOT TRAILERS: {:?}", trailers);
}
Ok(())
}
Hello, This works fine in http1 but in http2 it fails strangely.
Output
Cargo.toml
main.rs