AsyncAlgorithms: taming streams

One package, the missing vocabulary

The async-sequences lesson gave you the raw material: any source can become an AsyncSequence, and for await consumes it. What the standard library doesn't ship is the combinators — the things Combine veterans reach for on day one. They live in swift-async-algorithms, Apple's open-source package:

// Package.swift — the modules-spm lesson's mechanics:
.package(url: "https://github.com/apple/swift-async-algorithms", from: "1.0.0")

import AsyncAlgorithms

The catalogue, grouped the way you'll actually reach for it:

Note the shape of the combining family: merge(a, b) is a free function taking the sequences, not a method on one of them — none of the inputs is privileged. The time and shaping families are methods, because they transform a single upstream.

merge, zip, combineLatest — what does each wait for?

The three combiners differ only in their waiting policy, and every real screen maps cleanly onto one of them:

merge waits for nothing. Elements surface in arrival order, whoever produced them. One feed for the UI out of N independent sources — the pretest's fix:

Five quotes, interleaved purely by time. Delete every LSE element and the NYSE ones are untouched — independence is the property merge preserves and zip destroys.

zip waits for a full set — element N pairs with element N. combineLatest waits only until every input has produced once, then re-emits on every change, carrying the latest of the others. The distinction is exactly the difference between "pair the events" and "recompute from current values" — predict it before you run it:

Time operators: the ticker and the search box

The search box you already solved: the Combine-migration lesson built debounce from .task(id: query) + Task.sleep, where each keystroke's cancellation restarts the clock. The package spells the same contract as an operator on the query stream:

for await query in queryStream.debounce(for: .milliseconds(300)) {
    await search(query)      // fires only after 300ms of silence
}

Debounce's contract: a gap means done. Which is precisely why it is the wrong tool for the other hot stream in the building:

Throttle's semantics are worth owning precisely — one emission per window, and by default the latest value in the window (latest: true), because a ticker that renders stale prices at a calm rate would be worse than none:

Six ticks became three renders; the 101.09 dip and 101.15 blip inside the first window were superseded before anyone saw them. (Real throttle implementations differ in leading/trailing-edge details — the invariant to quote is at most one per interval, latest wins; with the package that's .throttle(for: .milliseconds(200)).)

The third time tool is the metronome: AsyncTimerSequence — for await over ticks, the structured replacement for a polling Timer (refresh the order book every second while the market's open, cancelled automatically when the consuming task dies).

Shaping: dedupe and batch

Two more operators carry their weight in any feed-driven app:

removeDuplicates() drops consecutive repeats — a quote stream that re-broadcasts an unchanged price shouldn't re-render the cell. Consecutive is the operative word: a price that returns later is a genuine change, which is why this is not a Set. (The first exercise makes you prove the difference.)

chunks(ofCount:) batches — 25 analytics events per upload instead of 25 uploads. And the fintech-shaped variant batches **by count or time**, whichever comes first, so a quiet stream still flushes:

// Ship every 25 events, or every 4 seconds, whichever comes first:
for await batch in events.chunked(by: .repeating(every: .seconds(4))) {
    try await upload(batch)
}

If that "batch, but flush on a deadline" shape feels familiar, it should — it's the offline-sync lesson's outbox drain policy, reborn as a stream operator.

The pipeline, end to end

Put the lesson together into the thing a trading app actually ships — socket to screen:

// 1. Bridge (the async-sequences lesson): socket callbacks -> AsyncStream
func quoteStream() -> AsyncStream<Quote> {
    AsyncStream(bufferingPolicy: .bufferingNewest(64)) { continuation in
        socket.onMessage = { continuation.yield(Quote(decoding: $0)) }   // the Codable lesson's boundary
        continuation.onTermination = { _ in socket.disconnect() }        // consumer died -> close the socket
    }
}

// 2. Shape it (this lesson):
let renderFeed = quoteStream()
    .removeDuplicates()                      // unchanged price -> no work
    .throttle(for: .milliseconds(250))       // humane rate, latest value

// 3. Consume it (the async-await lesson): lifetime owned by the view
.task {
    for await quote in renderFeed {
        model.apply(quote)                   // @MainActor by the actors lesson's rules
    }
}

Every line is a lesson you've already done — the bridge with its termination cleanup, the buffering policy (.bufferingNewest: a UI wants fresh, not history), the operators, the .task that cancels the whole pipeline when the view disappears, and the socket closing because of that cancellation, all the way down. That cascade — view gone → task cancelled → stream terminated → socket closed — is the answer to "how do you make sure the WebSocket doesn't leak", and it's four lessons shaking hands.