How to stream a response?

[alshdavid]

Hi all, I am trying to stream a response back to the client to implement server sent events and am wondering what the correct way to do this with hyper.

As a basic launch off point I have tried to implement a basic streamed text response where I can call write().await multiple times on it

Here's what I've tried:

use std::convert::Infallible;
use std::net::SocketAddr;

use futures::TryStreamExt;
use http_body_util::combinators::BoxBody;
use http_body_util::BodyExt;
use http_body_util::StreamBody;
use hyper::body::Bytes;
use hyper::body::Incoming;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::Request;
use hyper::Response;
use hyper_util::rt::TokioIo;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;

fn main() {
  tokio::runtime::Builder::new_multi_thread()
    .enable_all()
    .worker_threads(num_cpus::get_physical())
    .build()
    .unwrap()
    .block_on(async {
      let addr = SocketAddr::from(([127, 0, 0, 1], 8080));
      let listener = TcpListener::bind(addr).await.unwrap();

      loop {
        let (stream, _) = listener.accept().await.unwrap();
        let io = TokioIo::new(stream);

        tokio::task::spawn(async move {
          http1::Builder::new()
            .serve_connection(io, service_fn(root))
            .await
            .unwrap();
        });
      }
    });
}

async fn root(_request: Request<Incoming>) -> Result<Response<BoxBody<Bytes, std::io::Error>>, Infallible> {
  // My attempt at creating a reader/writer pair that I can write to asynchronously 
  let (mut writer, reader) = tokio::io::duplex(1024);
  
  let reader_stream = tokio_util::io::ReaderStream::new(reader);
  let stream_body = StreamBody::new(reader_stream.map_ok(hyper::body::Frame::data));
  let boxed_body = stream_body.boxed();
  
  let response = hyper::Response::builder()
    .header("Content-Type", "text/html")
    .status(hyper::StatusCode::OK)
    .body(boxed_body)
    .unwrap();

  tokio::task::spawn(async move {
    writer.write(b"hello world").await.unwrap(); // <- can call write() multiple times
  });

  Ok(response)
}

However this approach decimates performance. What's the correct approach to achieve this?

[seanmonstar]

How does it decimate performance? Compared to a single full body of "hello world"? I wouldn't worry about that. You're better off checking performance of your actual use case.

As for making a streaming writer, that looks fine. At least, if you want something that looks like an AsyncWrite. If not, you can use something like an mpsc::channel and send owned buffers. There's several streaming examples in https://github.com/hyperium/hyper/tree/master/examples