Struct StreamDownload

pub struct StreamDownload<P: StorageProvider> { /* private fields */ }

Represents content streamed from a remote source. This struct implements read and seek so it can be used as a generic source for libraries and applications that operate on these traits. On creation, an async task is spawned that will immediately start to download the remote content.

Any read attempts that request part of the stream that hasn’t been downloaded yet will block until the requested portion is reached. Any seek attempts that meet the same criteria will result in additional request to restart the stream download from the seek point.

If the stream download hasn’t completed when this struct is dropped, the task will be cancelled.

If the stream stalls for any reason, the download task will attempt to automatically reconnect. This reconnect interval can be controlled via Settings::retry_timeout. Server-side failures are not automatically handled and should be retried by the supplied SourceStream implementation if desired.

Implementations

impl<P: StorageProvider> StreamDownload<P>

pub async fn new_http( url: Url, storage_provider: P, settings: Settings<HttpStream<Client>>, ) -> Result<Self, StreamInitializationError<HttpStream<Client>>>

Available on crate feature reqwest only.

Creates a new StreamDownload that accesses an HTTP resource at the given URL.

Example

use std::error::Error;
use std::io::{self, Read};
use std::result::Result;

use stream_download::source::DecodeError;
use stream_download::storage::temp::TempStorageProvider;
use stream_download::{Settings, StreamDownload};

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let mut reader = match StreamDownload::new_http(
        "https://some-cool-url.com/some-file.mp3".parse()?,
        TempStorageProvider::default(),
        Settings::default(),
    )
    .await
    {
        Ok(reader) => reader,
        Err(e) => return Err(e.decode_error().await)?,
    };

    tokio::task::spawn_blocking(move || {
        let mut buf = Vec::new();
        reader.read_to_end(&mut buf)?;
        Ok::<_, io::Error>(())
    })
    .await??;
    Ok(())
}

pub async fn new_http_with_middleware( url: Url, storage_provider: P, settings: Settings<HttpStream<ClientWithMiddleware>>, ) -> Result<Self, StreamInitializationError<HttpStream<ClientWithMiddleware>>>

Available on crate feature reqwest-middleware only.

Creates a new StreamDownload that accesses an HTTP resource at the given URL. It uses the reqwest_middleware::ClientWithMiddleware client instead of the default reqwest client. Any global middleware set by Settings::add_default_middleware will be automatically applied.

Example

use std::error::Error;
use std::io::{self, Read};
use std::result::Result;

use reqwest_retry::RetryTransientMiddleware;
use reqwest_retry::policies::ExponentialBackoff;
use stream_download::source::DecodeError;
use stream_download::storage::temp::TempStorageProvider;
use stream_download::{Settings, StreamDownload};

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let retry_policy = ExponentialBackoff::builder().build_with_max_retries(3);
    Settings::add_default_middleware(RetryTransientMiddleware::new_with_policy(retry_policy));

    let mut reader = match StreamDownload::new_http_with_middleware(
        "https://some-cool-url.com/some-file.mp3".parse()?,
        TempStorageProvider::default(),
        Settings::default(),
    )
    .await
    {
        Ok(reader) => reader,
        Err(e) => return Err(e.decode_error().await)?,
    };

    tokio::task::spawn_blocking(move || {
        let mut buf = Vec::new();
        reader.read_to_end(&mut buf)?;
        Ok::<_, io::Error>(())
    })
    .await??;
    Ok(())
}

pub async fn new_async_read<T>( params: AsyncReadStreamParams<T>, storage_provider: P, settings: Settings<AsyncReadStream<T>>, ) -> Result<Self, StreamInitializationError<AsyncReadStream<T>>>

where T: AsyncRead + Send + Sync + Unpin + 'static,

Available on crate feature async-read only.

Creates a new StreamDownload that uses an AsyncRead resource.

Example reading from stdin

use std::error::Error;
use std::io::{self, Read};
use std::result::Result;

use stream_download::async_read::AsyncReadStreamParams;
use stream_download::storage::temp::TempStorageProvider;
use stream_download::{Settings, StreamDownload};

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let mut reader = StreamDownload::new_async_read(
        AsyncReadStreamParams::new(tokio::io::stdin()),
        TempStorageProvider::new(),
        Settings::default(),
    )
    .await?;

    tokio::task::spawn_blocking(move || {
        let mut buf = Vec::new();
        reader.read_to_end(&mut buf)?;
        Ok::<_, io::Error>(())
    })
    .await??;
    Ok(())
}

pub async fn new_process( params: ProcessStreamParams, storage_provider: P, settings: Settings<ProcessStream>, ) -> Result<Self, StreamInitializationError<ProcessStream>>

Available on crate feature process only.

Creates a new StreamDownload that uses a Command as input.

Example

use std::error::Error;
use std::io::{self, Read};
use std::result::Result;

use stream_download::process::{Command, ProcessStreamParams};
use stream_download::storage::temp::TempStorageProvider;
use stream_download::{Settings, StreamDownload};

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let mut reader = StreamDownload::new_process(
        ProcessStreamParams::new(Command::new("cat").args(["./assets/music.mp3"]))?,
        TempStorageProvider::new(),
        Settings::default(),
    )
    .await?;

    tokio::task::spawn_blocking(move || {
        let mut buf = Vec::new();
        reader.read_to_end(&mut buf)?;
        Ok::<_, io::Error>(())
    })
    .await??;
    Ok(())
}

pub async fn new<S>( params: S::Params, storage_provider: P, settings: Settings<S>, ) -> Result<Self, StreamInitializationError<S>>

where S: SourceStream, S::Error: Debug + Send,

Creates a new StreamDownload that accesses a remote resource at the given URL.

Example

use std::error::Error;
use std::io::{self, Read};
use std::result::Result;

use reqwest::Client;
use stream_download::http::HttpStream;
use stream_download::storage::temp::TempStorageProvider;
use stream_download::{Settings, StreamDownload};

use crate::stream_download::source::DecodeError;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let mut reader = match StreamDownload::new::<HttpStream<Client>>(
        "https://some-cool-url.com/some-file.mp3".parse()?,
        TempStorageProvider::default(),
        Settings::default(),
    )
    .await
    {
        Ok(reader) => reader,
        Err(e) => return Err(e.decode_error().await)?,
    };

    tokio::task::spawn_blocking(move || {
        let mut buf = Vec::new();
        reader.read_to_end(&mut buf)?;
        Ok::<_, io::Error>(())
    })
    .await??;
    Ok(())
}

pub async fn from_stream<S>( stream: S, storage_provider: P, settings: Settings<S>, ) -> Result<Self, StreamInitializationError<S>>

where S: SourceStream, S::Error: Debug + Send,

Creates a new StreamDownload from a SourceStream.

Example

use std::error::Error;
use std::io::Read;
use std::result::Result;

use reqwest::Client;
use stream_download::http::HttpStream;
use stream_download::storage::temp::TempStorageProvider;
use stream_download::{Settings, StreamDownload};

use crate::stream_download::source::DecodeError;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let stream = HttpStream::new(
        Client::new(),
        "https://some-cool-url.com/some-file.mp3".parse()?,
    )
    .await?;

    let mut reader = match StreamDownload::from_stream(
        stream,
        TempStorageProvider::default(),
        Settings::default(),
    )
    .await
    {
        Ok(reader) => reader,
        Err(e) => Err(e.decode_error().await)?,
    };
    Ok(())
}

pub fn cancel_download(&self)

Cancels the background task that’s downloading the stream content. This has no effect if the download is already completed.

pub fn cancellation_token(&self) -> CancellationToken

Returns the CancellationToken for the download task. This can be used to cancel the download task before it completes.

pub fn handle(&self) -> StreamHandle

Returns a StreamHandle that can be used to interact with the stream remotely.

pub fn content_length(&self) -> Option<u64>

Returns the content length of the stream, if available.

Trait Implementations

impl<P: Debug + StorageProvider> Debug for StreamDownload<P>

where P::Reader: Debug,

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

Formats the value using the given formatter.

impl<P: StorageProvider> Drop for StreamDownload<P>

fn drop(&mut self)

Executes the destructor for this type.

impl<P: StorageProvider> Read for StreamDownload<P>

fn read(&mut self, buf: &mut [u8]) -> Result<usize>

Pull some bytes from this source into the specified buffer, returning how many bytes were read.

fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> Result<usize, Error>

Like read, except that it reads into a slice of buffers.

fn is_read_vectored(&self) -> bool

🔬This is a nightly-only experimental API. (can_vector)

Determines if this Reader has an efficient read_vectored implementation.

fn read_to_end(&mut self, buf: &mut Vec<u8>) -> Result<usize, Error>

Reads all bytes until EOF in this source, placing them into buf.

fn read_to_string(&mut self, buf: &mut String) -> Result<usize, Error>

Reads all bytes until EOF in this source, appending them to buf.

fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), Error>

Reads the exact number of bytes required to fill buf.

fn read_buf(&mut self, buf: BorrowedCursor<'_>) -> Result<(), Error>

🔬This is a nightly-only experimental API. (read_buf)

Pull some bytes from this source into the specified buffer.

fn read_buf_exact(&mut self, cursor: BorrowedCursor<'_>) -> Result<(), Error>

🔬This is a nightly-only experimental API. (read_buf)

Reads the exact number of bytes required to fill cursor.

fn by_ref(&mut self) -> &mut Self

where Self: Sized,

Creates a “by reference” adaptor for this instance of Read.

fn bytes(self) -> Bytes<Self>

where Self: Sized,

Transforms this Read instance to an Iterator over its bytes.

fn chain<R>(self, next: R) -> Chain<Self, R>

where R: Read, Self: Sized,

Creates an adapter which will chain this stream with another.

fn take(self, limit: u64) -> Take<Self>

where Self: Sized,

Creates an adapter which will read at most limit bytes from it.

impl<P: StorageProvider> Seek for StreamDownload<P>

fn seek(&mut self, relative_position: SeekFrom) -> Result<u64>

Seek to an offset, in bytes, in a stream.

fn rewind(&mut self) -> Result<(), Error>

Rewind to the beginning of a stream.

fn stream_len(&mut self) -> Result<u64, Error>

🔬This is a nightly-only experimental API. (seek_stream_len)

Returns the length of this stream (in bytes).

fn stream_position(&mut self) -> Result<u64, Error>

Returns the current seek position from the start of the stream.

fn seek_relative(&mut self, offset: i64) -> Result<(), Error>

Seeks relative to the current position.