Struct Streamer 

pub struct Streamer<'a, P, F> { /* private fields */ }

Stream a single blob’s verified front to a consumer, paced by consumption.

Holds the injected pull machinery — the blob’s holders, the [SourceProvider] that builds their lanes, the command-wide [PeerHealth], and one funder — plus a scratch directory for the fill store. The front is fetched through [crate::acquire] across a small, bounded set of lanes; a lane that faults cools and returns. Streamer::open consumes it to start one stream.

Read inside StreamDrive::alongside, so the drive keeps paying and draining open legs while the reader waits on its consumer. See the stream example for the whole sequence.

use std::path::Path;
use std::sync::Arc;

use decdn_client::source::BlobSource;
use decdn_client::{
    Funder, NoCache, ProgressClock, PullConfig, StaticSources, StopPolicy, StreamCandidate,
    Streamer,
};

async fn stream<S: BlobSource, F: Funder>(
    candidates: Vec<StreamCandidate<S>>,
    funder: F,
    hash: [u8; 32],
    total_bytes: u64,
    scratch: &Path,
) -> anyhow::Result<()> {
    let streamer = Streamer::new(StaticSources::new(candidates)?, funder, scratch);
    let stop = StopPolicy::new(true, None, Arc::new(ProgressClock::new()));
    let (mut reader, mut drive) = streamer
        .open(hash, total_bytes, &PullConfig::new(), Arc::new(NoCache), stop)
        .await?;
    let mut out = tokio::io::stdout();
    drive.alongside(tokio::io::copy(&mut reader, &mut out)).await?;
    Ok(())
}

Implementations

impl<'a, P, F> Streamer<'a, P, F>

pub fn new(provider: P, funder: F, scratch: &'a Path) -> Self

Build a streamer whose holders and lanes provider finds and builds, funded through funder, filling into scratch.

The stream starts by discovering its holders, under a fresh RecoveryGate. The builder methods change both.

pub fn holders(self, holders: Vec<Holder>) -> Self

Start the stream from holders, in preference order, instead of discovering them. A probe’s measured holders go here.

pub fn recovery(self, recovery: Arc<RecoveryGate>) -> Self

Run the funding recovery step under recovery, the fetch’s gate. A caller that has already run part of the fetch under a gate (a first open, an earlier pass) passes that gate, so the progress rule spans the whole fetch.

pub fn credentials(self, slot: CredentialSlot) -> Self

Pay as a delegate under slot’s credential, as crate::Downloader::credentials does.

pub const fn max_blob_bytes(self, max_blob_bytes: u64) -> Self

Cap the stream at max_blob_bytes (0 for no cap). A size claim above the cap is clamped to it before the fill store is sized, and a blob that holds bytes past the cap ends the stream with crate::BlobTooLarge ([crate::AcquireEnv::max_blob_bytes]).

impl<P, F> Streamer<'_, P, F>

where P: SourceProvider, F: Funder,

pub async fn open<'a>( self, hash: [u8; 32], total_bytes: u64, config: &PullConfig, cache: Arc<dyn BlobCache>, stop: StopPolicy, ) -> Result<(VerifiedReader, StreamDrive<'a>)>

where P: 'a, F: 'a,

Open a consumption-paced stream over hash (whose content length is total_bytes), returning the VerifiedReader the caller drains and the StreamDrive that fetches into it.

The caller MUST poll the drive for as long as it reads — for example with StreamDrive::alongside around its read loop. The drive runs apart from the reads so a paid leg keeps paying while the consumer is slow or paused; the read-ahead window, not the reads, bounds it. Dropping the drive stops the fetch, and the reader then ends with an error after the verified bytes it already holds.

The drive borrows 'a from its sources — a real PeerSource over a borrowed Endpoint is not 'static — so the caller keeps the endpoint (and provider) alive for as long as it polls the drive. The reader owns its state and borrows nothing.

stop decides when a stream that makes no verified progress gives up; the reader then ends with crate::GaveUp’s message.

On a REVISIT — cache already holds the whole blob — the reader serves straight from the cache, the drive is already complete, and the network is never touched. Otherwise the fetch never runs more than config.read_ahead_bytes (raised to at least [PULL_WINDOW_FLOOR]) ahead of the consumer’s cursor, and the verified blob is teed to cache for the next revisit.

total_bytes is the first size claim, a hint. A resumed store’s bound wins, and a leg that verifies the final chunk proves the size. hash is the bao root every byte is verified against; a wrong hash surfaces as a read error, never as silent corruption.

Errors

A cache read fault at open, or a fill-store open/create I/O error. Faults DURING the fetch (a refused or stalled pull, a verification failure) surface later, from the reader.

Trait Implementations

impl<P, F> Debug for Streamer<'_, P, F>

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

Formats the value using the given formatter.