Using QUIC

impl Connection {
    async fn open_uni(&self) -> Result<SendStream, ConnectionError>;
    async fn accept_uni(&self) -> Result<Option<RecvStream>, ConnectionError>;
    
    async fn open_bi(&self) -> Result<(SendStream, RecvStream), ConnectionError>;
    async fn accept_bi(&self) -> Result<Option<(SendStream, RecvStream)>, ConnectionError>;
}

async fn connecting_endpoint(conn: Connection, request: &[u8]) -> Result<Vec<u8>> {
    let (mut send, mut recv) = conn.open_bi().await?;
    send.write_all(request).await?;
    send.finish()?;
    
    let response = recv.read_to_end(MAX_RESPONSE_SIZE).await?;
    
    conn.close(0u32.into(), b"I have everything, thanks!");
    
    Ok(response)
}

async fn accepting_endpoint(conn: Connection) -> Result<()> {
    let (mut send, mut recv) = conn.accept_bi().await?.ok_or_else(|| anyhow!("connection closed"))?;
    let request = recv.read_to_end(MAX_REQUEST_SIZE).await?;
    
    let response = compute_response(&request);
    send.write_all(&response).await?;
    send.finish()?;
    
    conn.closed().await;
    
    Ok(())
}

async fn connecting_endpoint(conn: Connection, mut request: impl Stream<Item = u64>) -> Result<()> {
    let (mut send, mut recv) = conn.open_bi().await?;
    
    // concurrently read the responses
    let read_task = tokio::spawn(async move {
	    let mut buf = [0u8; size_of::<u64>()];
	    // read_exact will return `Err` once the other side
	    // finishes its stream
        while recv.read_exact(&mut buf).await.is_ok() {
	        let number = u64::from_be_bytes(buf);
            println!("Read response: {number}");
        }
    });
    
    while let Some(number) = request.next().await {
        let bytes = number.to_be_bytes();
        send.write_all(&bytes).await?;
    }
    send.finish()?;
    
    // we close the connection after having read all data
    read_task.await?;
    conn.close(0u32.into(), b"done");
    
    Ok(())
}

async fn accepting_endpoint(conn: Connection) -> Result<()> {
    let (mut send, mut recv) = conn.accept_bi().await?.ok_or_else(|| anyhow!("connection closed"))?;
    
    let mut buf = [0u8; size_of::<u64>()];
    while recv.read_exact(&mut buf).await.is_ok() {
	    let number = u64::from_be_bytes(buf);
	    let doubled = number.wrapping_mul(2).to_be_bytes();
	    send.write_all(&doubled).await?;
    }
    send.finish()?;
    
	// the other side will tell us when it's done reading our data
	conn.closed().await;
	
	Ok(())
}

// The connecting endpoint can call this multiple times
// for one connection.
// When it doesn't want to do more requests and has all
// responses, it can close the connection.
async fn request(conn: &Connection, request: &[u8]) -> Result<Vec<u8>> {
    let (mut send, mut recv) = conn.open_bi().await?;
    send.write_all(request).await?;
    send.finish()?;
    
    let response = recv.read_to_end(MAX_RESPONSE_SIZE).await?;
    
    Ok(response);
}

// The accepting endpoint will call this to handle all
// incoming requests on a single connection.
async fn handle_requests(conn: Connection) -> Result<()> {
    loop {
        let stream = conn.accept_bi().await?;
        match stream {
            Some((send, recv)) => {
                tokio::spawn(handle_request(send, recv));
            }
            None => break, // connection closed
        }
    }
    Ok(())
}

async fn handle_request(mut send: SendStream, mut recv: RecvStream) -> Result<()> {
    let request = recv.read_to_end(MAX_REQUEST_SIZE).await?;
    
    let response = compute_response(&request);
    send.write_all(&response).await?;
    send.finish()?;
    
    Ok(())
}

async fn connecting_endpoint(conn: Connection, mut notifications: impl Stream<Item = Bytes> + Unpin) -> Result<()> {
    let mut send = conn.open_uni().await?;
    
    let mut send_frame = LengthDelimitedCodec::builder().new_write(send);
    while let Some(notification) = notifications.next().await {
        send_frame.send(notification).await?;
    }
    
    send_frame.get_mut().finish()?;
    conn.closed().await;
    
    Ok(())
}

async fn accepting_endpoint(conn: Connection) -> Result<()> {
    let recv = conn.accept_uni().await?.ok_or_else(|| anyhow!("connection closed"))?;
    let mut recv_frame = LengthDelimitedCodec::builder().new_read(recv);
    
    while let Some(notification) = recv_frame.try_next().await? {
        println!("Received notification: {notification:?}");
    }
    
    conn.close(0u32.into(), b"got everything!");
    
    Ok(())
}

async fn connecting_endpoint(conn: Connection, request: &[u8]) -> Result<()> {
    let (mut send, recv) = conn.open_bi().await?;
	send.write_all(request).await?;
	send.finish()?;
	
	let mut recv_frame = LengthDelimitedCodec::builder().new_read(recv);
	while let Some(response) = recv_frame.try_next().await? {
	    println!("Received response: {response:?}");
	}
	
	conn.close(0u32.into(), b"thank you!");
    
    Ok(())
}

async fn accepting_endpoint(conn: Connection) -> Result<()> {
	let (send, mut recv) = conn.accept_bi().await?.ok_or_else(|| anyhow!("connection closed"))?;
	let request = recv.read_to_end(MAX_REQUEST_SIZE).await?;
	
	let mut send_frame = LengthDelimitedCodec::builder().new_write(send);
	let mut responses = responses_for_request(&request);
	while let Some(response) = responses.next().await {
	    send_frame.send(response).await?;
	}
	send_frame.get_mut().finish()?;

	conn.closed().await;
	
	Ok(())
}

fn responses_for_request(req: &[u8]) -> impl Stream<Item = Bytes> {
    // ...
}

async fn connnecting_side(conn: Connection, request: &[u8]) -> Result<()> {
    let mut send = conn.open_uni().await?;
    send.write_all(request).await?;
    send.finish()?;
    
    let recv_tasks = TaskTracker::new();
    // accept_uni will return `Ok(None)` once the connection is closed
    loop {
        match conn.accept_uni().await? {
            Some(recv) => {
                recv_tasks.spawn(handle_response(recv));
            }
            None => break,
        }
    }
	recv_tasks.wait().await;
	conn.close(0u32.into(), b"Thank you!");
	
	Ok(())
}

async fn handle_connection(conn: Connection) -> Result<()> {
    futures_lite::future::race(
		run_protocol(conn.clone()),
		async move {
			conn.closed().await;
			anyhow::Ok(())
		},
    ).await?;
    Ok(())
}

async fn run_protocol(conn: Connection) -> Result<()> {
	// run normal protocol flow
	// once we realize we want to abort the connection flow
	conn.close(0u32.into(), b"ah sorry, have to go!");
	
	Ok(())
}

async fn accepting_endpoint(conn: Connection) -> Result<()> {
    let (mut send, mut recv) = conn.accept_bi().await?.ok_or_else(|| anyhow!("connection closed"))?;
    let request = recv.read_to_end(MAX_REQUEST_SIZE).await?;
    
    // We can start sending the response immediately without waiting
    // for the client to finish the handshake
    let response = compute_response(&request);
    send.write_all(&response).await?;
    send.finish()?;
    
    conn.closed().await;
    Ok(())
}