Streams

ReadableStream is everywhere. CompressionStream - which the transforms need - is newer, so it has its own check. During prerender/SSR both return false rather than throwing, so defer them to OnAfterRenderAsync.

@inject Bit.Butil.Streams streams

var supported = await streams.IsSupported();
var transforms = await streams.IsTransformSupported();

The difference between 'download it, then look at it' and 'look at it as it downloads': a gigabyte through Fetch.Send is a gigabyte of managed memory, while a gigabyte through here is one chunk at a time. TotalBytes is the Content-Length header, which describes the transfer rather than the stream - a compressed response reports the encoded size while the chunks you read are the decoded ones, and a chunked response sends no header at all. It is a hint for a progress bar, never the number of bytes you will read. The final read carries no data, so the loop ends on Done rather than on an empty array.

@implements IAsyncDisposable
@inject Bit.Butil.Streams streams

@code {
    private async Task Read()
    {
        var response = await streams.FromResponse(new FetchRequest { Url = "/api/stream" });
        if (response.Stream is null) { /* response.Error says why */ return; }

        // The stream holds a live connection open, so it is disposed however the loop ends.
        await using var stream = response.Stream;

        var received = 0L;
        while (true)
        {
            var chunk = await stream.Read();
            if (chunk.Done) break;
            received += chunk.Data!.Length;   // ... and do something with it now
        }
    }
}

// Something worth streaming: a body that arrives in visible instalments rather than all at once,
// with a Content-Length so a progress bar has a denominator. The pause between chunks is what makes
// "read it as it arrives" observably different from "wait, then read it".
app.MapGet("/api/stream", async (HttpContext context, CancellationToken cancellationToken) =>
{
    const int chunks = 20;
    const int chunkSize = 4096;

    context.Response.Headers.ContentType = "application/octet-stream";
    context.Response.ContentLength = chunks * chunkSize;

    var payload = new byte[chunkSize];
    Array.Fill(payload, (byte)'x');

    for (var i = 0; i < chunks; i++)
    {
        await context.Response.Body.WriteAsync(payload, cancellationToken);
        // Without the flush the server buffers the whole body and the client sees one chunk.
        await context.Response.Body.FlushAsync(cancellationToken);
        await Task.Delay(100, cancellationToken);
    }
});

Cancelling a stream tells the source to give up - for a fetch body that aborts the download rather than letting it finish into nothing. Start this section's read and press Cancel; the server stops sending, and the read loop ends where it was.

await response.Stream!.Cancel("the user navigated away");

// the next Read answers Done, and the request is gone from the network panel

Tee splits a stream in two, each getting every chunk - for hashing a download while also writing it somewhere, say. The original handle is spent: the stream has been split, not copied. Both branches are fed from the one source, so a branch nobody reads makes the browser buffer for it indefinitely - read both, or cancel the one you do not want.

var split = await response.Stream!.Tee();
var (first, second) = split!.Value;

// both see every chunk; the original handle is spent
var counting = CountBytes(first);
var hashing = HashBytes(second);
await Task.WhenAll(counting, hashing);

The transforms are the browser's own codecs, and PipeThrough is what accepts them. PipeTo runs the whole pipe in one call - back-pressure, closing the destination, and propagating an error from either end all happen inside it. The sink's callback is where the bytes reach C#, and the producer waits for it to return before sending the next chunk, so a slow handler slows the download rather than queueing behind it.

var gzip = await streams.CreateCompression(CompressionFormat.Gzip);
var sink = await streams.CreateWritable(
    onChunk: bytes => { compressed += bytes.Length; },
    onFinished: reason => { /* null on a clean close */ });

var compressedStream = await response.Stream!.PipeThrough(gzip!);
var error = await compressedStream!.PipeTo(sink!);   // null on success

A writable does not have to be fed by a pipe. Write waits until the sink is ready for the chunk rather than queueing without limit, which is what makes a producer in a loop run at the speed of its consumer. A stream being piped into cannot also be written to by hand - the pipe holds it.

var sink = await streams.CreateWritable(bytes => received.AddRange(bytes));

foreach (var chunk in chunks)
{
    await sink!.Write(chunk);   // returns when the sink is ready for the next one
}
await sink!.Close();