Composing Streams: Resumable and FlatMap

The two most sophisticated decoders in the framework — auto-reconnecting resumable streams and flat-map over a stream of streams — are the proof that the decoder layer composes. Both are how the icecast_connect client stays connected to an internet radio mount for hours across network hiccups. The lesson underneath: you write synchronous policy; the framework owns the async machinery.

Reconnection lives at the decoder layer on purpose. It is the only layer that persists for the whole life of stream consumption: a pipeline member returns before the body streams, and a runner returns once the output handle is produced. Only the decoder knows how far the stream got — the byte offset, the last event id, the candidate it was on — and therefore how to resume without gaps or duplicates.

Resumable streams

A Resumable stream reconnects itself when the connection breaks, transparently, so the consumer sees one uninterrupted ResumableIterator<Item> — a StreamIterator whose metadata slot is closed and whose items stay QueryError-fallible, because the stream issues follow-up requests. You implement two traits and a synchronous policy; the framework runs the reconnect manager.

• DecodeResumable is the connection decode: a Parser (a StreamParser for the framing), an Item, parser(headers, config), and decode_item. It is implemented on the initial endpoint and on its resume endpoint, returning the same decode_item, so the decode is defined once and shared. A blanket gives every such endpoint a plain streaming decoder, StreamFnDecoder, so a resume endpoint is a genuine, independently callable streaming endpoint.
• Resumable adds the reconnect policy: the Cursor and ResumeEndpoint types, and four all-synchronous hooks.

icecast’s GetStream is the shipped model. Its cursor is a Failover — the index into the playlist’s candidate mounts — and its on_break decides three layers at once:

impl Resumable for GetStream {
    type ResumeEndpoint = Self;
    type Cursor = Failover;

    fn on_break(&self, ctx: BreakCtx<'_>, cursor: &mut Self::Cursor, rng: &ApiRng) -> Resume<Self> {
        let policy = self.reconnect.clone().unwrap_or_default();
        let candidates = self.candidates();

        match policy.decide(ctx.end, ctx.attempt) {
            ReconnectDecision::Stop => Resume::Stop,
            // Layer 1: rejoin the current candidate with the policy's backoff.
            ReconnectDecision::Reconnect => Resume::reconnect(self.at(&candidates, cursor.index))
                .after(policy.backoff(ctx.attempt, rng)),
            ReconnectDecision::Fail => {
                // Same-mount retries are spent. Layer 2: fail over to the next
                // playlist candidate, if any, before surfacing the failure.
                if cursor.index + 1 < candidates.len() {
                    cursor.index += 1;
                    Resume::reconnect(self.at(&candidates, cursor.index))   // a fresh mount: no backoff
                } else {
                    match ctx.end {
                        StreamEnd::Eof => Resume::Stop,   // every candidate spent on a clean end
                        _ => Resume::Fail,                // anything else surfaces the error
                    }
                }
            }
        }
    }

    fn idle_timeout(&self) -> Option<Duration> {
        self.idle_timeout
    }
}
// type Decoder = ResumableStreamDecoder;

on_break is the unified hook, called for every terminal. It receives a BreakCtx — the StreamProgress so far, attempt (reconnects already tried for this break, 0 the first time; it resets on a successful reconnect), and end, which distinguishes a clean EOF, a mid-stream error, a truncation, and a failed reconnect attempt — plus the mutable cursor and the request’s ApiRng, for jittered backoff. It returns a Resume: Resume::reconnect(endpoint), optionally .after(delay) and .keep_buffer(); Resume::Stop for a clean end; or Resume::Fail to surface the error. The partial parse buffer is discarded on a reconnect unless keep_buffer() says otherwise — the right default for rejoining a live feed or resuming at a byte offset, where the bytes before the break are either replayed by the server or gone.

The other hooks fill the cursor and bound the loop:

• observe(cursor, frame) updates the cursor from each parsed frame, before decoding filters it — a last event id, a response id and sequence number.
• on_connect(cursor, headers) updates it from each connection’s response head, on the initial connection and after every successful reconnect — a server-assigned resumption token, a Range acknowledgement.
• validate_continuation(..) (on DecodeResumable) checks that a resumed response answers this request — a 206 Partial Content continuing at the right offset — and routes a rejection through decode_error like an admission failure.
• max_reconnects() is the framework’s only built-in ceiling, on successful reconnects; it defaults to None, so a policy that always returns Reconnect is an infinite loop by the implementor’s own choice, and ctx.attempt is the cap on consecutive failed ones.
• idle_timeout() treats a silent connection as a break. Async only: a synchronous stream blocks inside one read, which no timer can interrupt.

A fired cancel token on the request’s context is terminal. Within a connection the transport is the cancel point; between connections the driver is — a break under a fired token ends the stream ahead of the policy, and a backoff wait is cut short. The policy’s verdict is never sought for work the caller has cancelled. Reconnects themselves are observable without being interleaved into the item stream: each one is a ReconnectNotification (with its ReconnectPhase) on the context’s notification channel, which bubbles to a subscriber on the base context or a per-request sender.

Custom stream parsers

Behind a resumable (or any) stream is a StreamParser — a state machine that turns a byte buffer into frames. The rules that trip people up:

fn parse(&mut self, buffer: &mut Vec<u8>) -> Result<Option<Item>, Error>;
fn flush_at_eof(&mut self, buffer: &mut Vec<u8>) -> Result<Option<Item>, Error> { Ok(None) }

Ok(None) means “need more bytes”, not EOF. The engine calls parse repeatedly: Ok(Some(item)) drains one frame, Ok(None) says “I’ve consumed what I can, feed me more”. End-of-stream is signalled by the transport, and then the engine calls flush_at_eof once, for a parser whose last frame may be unterminated — an SSE event with no trailing blank line, an NDJSON line with no final newline. icecast’s IcyParser returns Ok(None) when the buffer is empty or a metadata block isn’t fully buffered yet, and reports is_incomplete() == false because a partial trailing chunk at EOF is normal for a radio stream, never a truncation error. A parser that needs no framing at all — a raw byte relay — is PassthroughParser.

When a plain parser isn’t enough, the escalation ladder is: StreamParser (custom framing) → StreamFnDecoder (the blanket that turns any DecodeResumable into a one-shot streaming decoder) → type Decoder = Self (full control: impl ResponseDecoder<Self> directly on the endpoint, receiving the raw response and the door — the decoder system in depth). Reach up the ladder only as far as you must. Bytestream is read-once: a parser consumes its buffer and cannot rewind.

FlatMap: a stream of streams

FlatMapDecode flattens a two-level stream: an outer source yields items, and each item is turned into an inner endpoint whose stream is spliced into the output — a FlatMapIterator<Item>, the second stream handle with a contract of its own. icecast’s “listen to a playlist” is the archetype: the outer is a looping playlist pager, each entry becomes a bounded GetStream, and the flattened result is one continuous StreamEvent stream:

impl FlatMapDecode for ListenPlaylist {
    type OuterItem = StreamCandidate;
    type Item = StreamEvent;
    type Outer = Paged<RefetchPlaylist>;   // a looping outer: every page re-fetches the playlist
    type Inner = GetStream;                // queried per candidate

    fn outer(&self) -> Paged<RefetchPlaylist> {
        RefetchPlaylist { url: self.url.clone(), include_auth: self.include_auth }.paged()
    }

    fn inner(&self, candidate: &StreamCandidate) -> GetStream {
        // One bounded pass per candidate: connect once, stream until it ends, then let
        // the flatten fail over. Resilience comes from the failover + re-resolve loop.
        let mut endpoint = GetStream::new(candidate.url.clone());
        endpoint.reconnect = Some(ReconnectPolicy::never());
        endpoint.include_auth = self.include_auth;
        endpoint.idle_timeout = self.idle_timeout;
        endpoint
    }

    fn on_end(&self, brk: FlatMapBreak<'_, StreamCandidate>) -> Recovery {
        self.refetch.decide(&brk)
    }
}

Inner is any pass-through IntoEndpoint — a plain endpoint, or a derived input whose conversion can fail; the driver converts each outer item eagerly, and a failed conversion reaches on_end as SourceEnd::InnerUnavailable.

on_end runs whenever a source ends, clean completion included: an inner stream that reached its natural end (InnerCompleted) is as much a decision point as one that failed (InnerFailed), could not be built or sent (InnerUnavailable), or whose outer re-fetch failed (OuterFailed). Outer exhaustion is not a source end — when the outer iterator runs out, the flattened stream simply ends, and looping is the job of a self-refreshing outer such as the paged one above. The FlatMapBreak it receives answers end(), error() (when the source failed), item() (the outer item the inner came from), and the consecutive-failure count. It returns a Recovery:

• Continue { after } — advance to the next source, swallowing any error, optionally after a delay;
• Report { after } — emit the error as an item, then advance;
• Stop — emit the error as the final item, then end.

The default is policy::fail_fast — stop on the first error of any kind, advance on a clean end — and policy::skip_inner_stop_outer is the natural playlist policy: a dead track is skipped and reported, a dead playlist ends the stream. Inner queries derive their context from the originating request’s, so cancelling the flattened stream reaches the inner connections.

Composition is the payoff

The power is that these nest. In the listen example there are three layers of resilience, each owning its concern, and each a small synchronous decision:

• Layer 1 — same-mount rejoin: GetStream’s own on_break reconnects to the candidate it was on, with the policy’s backoff — when the endpoint is used on its own. Under the flatten, inner sets ReconnectPolicy::never(), so a candidate is one bounded pass.
• Layer 2 — candidate failover: the same on_break, out of same-mount retries, bumps the Failover cursor to the next mount in the playlist and reconnects there at once.
• Layer 3 — playlist re-resolve: on_end’s RefetchPolicy, with deterministic backoff, decides when the looping outer fetches the playlist again for a fresh candidate list.

The framework composes them into one StreamIterator that survives a dropped packet, an expired stream URL, and a rotated playlist without the consumer writing a line of reconnect code. A resumable download — Range at the last byte offset, If-Range on the validator — is the same Resumable machinery packaged as ResumeDownloadDecoder, in the streaming chapter. That’s the whole argument for putting reconnection at the decoder layer.

Next: Batching — several endpoints in one multipart/mixed request, each decoded by its own decoder.