stream body

[piquu]

Summary

im trying to stream the body but everything i can find about it is from a few years ago and outdated

async fn stream() -> impl IntoResponse {
    let (tx, mut rx) = channel::<Bytes>();

    tokio::spawn(async move {
        for i in 1..100 {
            tx.send(Bytes::from(format!("number: {i}")));
            tokio::time::sleep(Duration::from_millis(500)).await;
        }
    });

    Body::from_stream(rx)
}

axum version

0.7.7

[mladedav]

When asking for help, you might want to also include the error message.

I don't know what channel you're using but the tokio ones don't implement Stream. There are wrappers in tokio-stream like this one. Other streams have similar wappers or you can create one yourself.

[mladedav]

Use the tokio channel and wrap it in the struct I linked in the previous answer. That implements Stream while the receiver itself does not.

[piquu]

now it compiles but i get a error at runtime

pub async fn stream() -> impl IntoResponse {
    let (tx, rx) = tokio::sync::mpsc::channel::<Result<axum::body::Bytes, anyhow::Error>>(usize::MAX);

    tokio::spawn(async move {
        for i in 1..100 {
            tx.send(Ok(axum::body::Bytes::from(format!("number: {i}")))).await;
            tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        }
    });

    Body::from_stream(tokio_stream::wrappers::ReceiverStream::new(rx))
}

2024-10-25T06:33:17.696830Z DEBUG request{method=GET uri=/stream version=HTTP/1.1}: tower_http::trace::on_request: started processing request
thread 'tokio-runtime-worker' panicked at C:\Users\piqu\.cargo\registry\src\index.crates.io-6f17d22bba15001f\tokio-1.40.0\src\sync\batch_semaphore.rs:141:9:
a semaphore may not have more than MAX_PERMITS permits (2305843009213693951)

[mladedav]

I believe part of the message should have also been note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace. You need to use that to figure out which function you call is panicking and then read the documentation for that function. I'm sorry, but you have to do these steps yourself.

[piquu]

i got it working. the issue was that i set the channel buffer to usize::MAX. tysm for helping me

pub async fn stream() -> impl IntoResponse {
    let (tx, rx) = tokio::sync::mpsc::channel::<Result<axum::body::Bytes, anyhow::Error>>(100);

    tokio::spawn(async move {
        for i in 1..100 {
            tx.send(Ok(axum::body::Bytes::from(format!("number: {i}")))).await;
            tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        }
    });

    Body::from_stream(tokio_stream::wrappers::ReceiverStream::new(rx))
}