Struct ReaderStream

pub struct ReaderStream<S: Stream> { /* private fields */ }

An async reader that reads from a stream of async readers.

A ReaderStream can be constructed from any Stream of items of type T where T: AsyncRead, or from a single AsyncRead type using ReaderStream::one(). The AsyncRead implementation of ReaderStream progresses the stream forward, returning the contents of the inner readers. Thus, a ReaderStream can be thought of as a flattening of async readers.

ReaderStream is designed to be used as a building-block for stream-based responders by acting as the streamed_body of a Response, though it may also be used as a responder itself.

use std::io::Cursor;

use rocket::{Request, Response};
use rocket::futures::stream::{Stream, StreamExt};
use rocket::response::{self, Responder, stream::ReaderStream};
use rocket::http::ContentType;

struct MyStream<S>(S);

impl<'r, S: Stream<Item = String>> Responder<'r, 'r> for MyStream<S>
    where S: Send + 'r
{
    fn respond_to(self, _: &'r Request<'_>) -> response::Result<'r> {
        Response::build()
            .header(ContentType::Text)
            .streamed_body(ReaderStream::from(self.0.map(Cursor::new)))
            .ok()
    }
}

Responder

ReaderStream is a (potentially infinite) responder. No Content-Type is set. The body is unsized, and values are sent as soon as they are yielded by the internal stream.

Example

use rocket::response::stream::ReaderStream;
use rocket::futures::stream::{repeat, StreamExt};
use rocket::tokio::time::{self, Duration};
use rocket::tokio::fs::File;

// Stream the contents of `safe/path` followed by `another/safe/path`.
#[get("/reader/stream")]
fn stream() -> ReaderStream![File] {
    ReaderStream! {
        let paths = &["safe/path", "another/safe/path"];
        for path in paths {
            if let Ok(file) = File::open(path).await {
                yield file;
            }
        }
    }
}

// Stream the contents of the file `safe/path`. This is identical to
// returning `File` directly; Rocket responders stream and never buffer.
#[get("/reader/stream/one")]
async fn stream_one() -> std::io::Result<ReaderStream![File]> {
    let file = File::open("safe/path").await?;
    Ok(ReaderStream::one(file))
}

The syntax of ReaderStream! as an expression is identical to that of stream!.

Implementations

impl<R: Unpin> ReaderStream<One<R>>

pub fn one(reader: R) -> Self

Create a ReaderStream that yields exactly one reader, streaming the contents of the reader itself.

Example

Stream the bytes from a remote TCP connection:

use std::io;
use std::net::SocketAddr;

use rocket::tokio::net::TcpStream;
use rocket::response::stream::ReaderStream;

#[get("/stream")]
async fn stream() -> io::Result<ReaderStream![TcpStream]> {
    let addr = SocketAddr::from(([127, 0, 0, 1], 9999));
    let stream = TcpStream::connect(addr).await?;
    Ok(ReaderStream::one(stream))
}

Trait Implementations

impl<S: Stream> AsyncRead for ReaderStream<S>

where S::Item: AsyncRead + Send,

fn poll_read( self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>, ) -> Poll<Result<()>>

Attempts to read from the AsyncRead into buf.

impl<S: Stream + Debug> Debug for ReaderStream<S>

where S::Item: Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<S: Stream> From<S> for ReaderStream<S>

fn from(stream: S) -> Self

Converts to this type from the input type.

impl<'r, S> Responder<'r, 'r> for ReaderStream<S>

where S: Send + 'r + Stream, S::Item: AsyncRead + Send,

fn respond_to(self, _: &'r Request<'_>) -> Result<'r>

Returns Ok if a Response could be generated successfully. Otherwise, returns an Err with a failing Status.

impl<'__pin, S: Stream> Unpin for ReaderStream<S>

where PinnedFieldsOf<__Origin<'__pin, S>>: Unpin,