How to set Content-Type for ReaderStream?

[wangjia184]

pub(crate) async fn xxxxxxxxxxxxxxxxx(
    device_path: &str,
) -> ReaderStream![Cursor<Bytes>] {

    let mut rx : tokio::sync::mpsc::Receiver<Bytes> = // ...;

    ReaderStream! {
        while let Some(chunk) = rx.recv().await {
            yield Cursor::new(chunk);
        }
    }
}

I have the above code and it works well.
Now I want to add Content-Type for the response. how can I do that?

I tried the following code , but failed to connect

pub(crate) struct BinaryStream<S>(S);

impl<'r, S> Responder<'r, 'r> for BinaryStream<S>
where
    S: tokio::io::AsyncRead + Send + 'r,
    S: Unpin, // Make sure to add Unpin if needed
{
    fn respond_to(self, _: &'r Request<'_>) -> rocket::response::Result<'r> {
        Response::build()
            .header(ContentType::Binary)
            .streamed_body(self.0)
            .ok()
    }
}

[the10thWiz]

The following struct can be used to wrap any responder type, and sets the content_type to binary.

#[derive(Responder)]
#[response(content_type = "binary")]
pub(crate) struct BinaryStream<S>(S);

[wangjia184]

Thanks @the10thWiz
As you can see header(ContentType::Binary) is already there.
My question is that : how to connect tokio::sync::mpsc::Receiver to BinaryStream so that Content-Type will be added.

tokio::sync::mpsc::Receiver does not implement AsyncRead, is there anyway to connect them?

[wangjia184]

impl HttpFlvStream {
    pub fn new(receiver: tokio::sync::mpsc::Receiver<Bytes>) -> Self {
        Self {
            receiver,
            buffer: None,
        }
    }
}

// Implement AsyncRead for HttpFlvStream
impl tokio::io::AsyncRead for HttpFlvStream {
    fn poll_read(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        use std::pin::Pin;
        use std::task::Poll;
        // If we have leftover data in the buffer, read from it first
        if let Some(cursor) = self.buffer.as_mut() {
            let available_data = cursor.get_ref().len() as u64 - cursor.position();
            if available_data > 0 {
                let to_read = buf.remaining().min(available_data as usize);
                let start = cursor.position() as usize;
                buf.put_slice(&cursor.get_ref()[start..start + to_read]);
                cursor.set_position(cursor.position() + to_read as u64);
                return Poll::Ready(Ok(()));
            }
        }

        // If there's no data left, poll the receiver for the next message
        match Pin::new(&mut self.receiver).poll_recv(cx) {
            Poll::Ready(Some(bytes)) => {
                // Store the received data in the buffer and reset cursor
                self.buffer = Some(Cursor::new(bytes));
                self.poll_read(cx, buf) // Recurse to handle the buffer reading
            }
            Poll::Ready(None) => Poll::Ready(Ok(())), // Channel closed
            Poll::Pending => Poll::Pending,
        }
    }
}

// Implement Responder for HttpFlvStream
impl<'r> Responder<'r, 'r> for HttpFlvStream {
    fn respond_to(self, _: &'r Request<'_>) -> rocket::response::Result<'r> {
        Response::build()
            .header(ContentType::Binary)
            .streamed_body(self)
            .ok()
    }
}