How to cancel Stream when client disconnect.

[terrencehan]

Hi，I am new to Rustlang and Axum, and still struggling with the learning curve. Any help would be appreciated.

I am trying to use Axum to write a HTTP API that subscribes to Redis channels. So I used StreamBody to response with a stream to the client. Here’s what it looks like:

use axum::{
    routing::get,
    Router, body::StreamBody,
};
use futures_util::Stream;
use std::{net::{Ipv4Addr, SocketAddr}, pin::Pin, task::{Poll, Context}, thread, time::Duration, io};

#[tokio::main]
async fn main() {

    let app = Router::new()
        .route(
            "/stream",
            get(mystream),
        );

    let addr = SocketAddr::from((Ipv4Addr::UNSPECIFIED, 3000));
    tracing::debug!("listening on {}", addr);
    axum::Server::bind(&addr)
        .serve(app.into_make_service())
        .await
        .unwrap();
}

struct MyStream;

impl Stream for MyStream {
    type Item = io::Result<String>;

    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        println!("poll next");
        thread::sleep(Duration::from_secs(1));
        Poll::Ready(Some(Ok("1\n".to_string())))
    }
}

async fn mystream() -> StreamBody<impl Stream <Item = io::Result<String>>> {
    StreamBody::new(MyStream{})
}

These are my questions:

1. How do I end the stream? I noticed that even when the client disconnects (Ctrl+C with curl), poll_next is still called and prints poll next.
2. How do I respond immediately when a string with newline is supplied? Now, it seems to be buffered for 16 elements.”

Here’s a runnable demo: https://github.com/terrencehan/axum_streambody_example

Thanks

[terrencehan]

I'll try sse package first.

[daheige]

please see this demo: https://github.com/tokio-rs/axum/blob/main/examples/sse/src/main.rs