Channel streams put a stream between two tasks in one process. This page puts one between two processes: typed values cross a connection as framed transfers, and — the part that makes it a stream rather than a byte protocol — the way the stream ends crosses the wire too. A producer that finishes cleanly ends your loop; a producer that fails hands your on fail its reason; and a frame the reader does not speak refuses before a single value byte is read. One failure channel, one consumption form, whatever went wrong.

The example is examples/NetValueStream. Build and run it:

make -C mica-container/examples/NetValueStream run
make -C mica-container/examples/NetValueStream run MICA_EXTRA_FLAGS="--tasking multicore,2"

Two verbs and one closing frame

The writing side is two verbs from the net unit. WriteValue frames a plain value — an 8-byte versioned header, then the value’s bytes whole — and the stream’s end is one closing frame: WriteEnd for the clean end, WriteFail(address c, reason) to end it failing, with one word of reason.

for k := 1 to 3 do
begin
    served.id := k;
    served.price := 1.5 * (k as float64);
    WriteValue(address c, served) on fail continue;
end;

if mode = FeedFailing then
    WriteFail(address c, ReasonHalt) on fail continue
else
    WriteEnd(address c) on fail continue;

The reason word is the whole protocol between the two processes: the wire does not interpret it, the two ends agree on what it means. That is the same one-word closing a task-fed ring latches in process — the wire form of a fact the language already taught.

The reader is a generator

The reading side declares the stream it answers, failure domain included — the stream of T fails E form — and reads frames with ReadInto until the closing frame ends the loop:

generator Prices(c : Connection) : stream of Quote fails NetError;
var
    next : Quote;
    more : bool;
    os : OsError;
begin
    more := True;

    while more do
    begin
        more := ReadInto(address c, address next)
        on fail os do
        begin
            if os = Einval then
                fail Corrupt;

            fail PeerClosed;
        end;

        if more then
            emit next;
    end;

    if ClosingReason(c) = ReasonHalt then
        fail PeerClosed;
end;

Three different endings funnel into the one declared domain. The peer’s polite failure arrives as the closing reason and becomes PeerClosed at the last line. A transport death — the peer vanished mid-frame — surfaces as ReadInto’s typed failure and is lifted into the same constructor. And a foreign frame (a version byte this reader does not speak) fails as Einval before any value byte is read, and becomes Corrupt. The consumer cannot tell a polite failure from a lifted one except by its constructor — which is the design: the wire’s troubles are the domain’s constructors, not a second error system.

Note what the generator’s parameter is: a Connection by value. A generator’s inputs are deep-copied into the instance at open, so the generator owns its read cursors for the stream’s whole life and no other task can name them.

Walking it

The consumer is a for loop with the same on fail every call site writes:

for item in Prices(client) do
begin
    total := total + item.price;
end
on fail reason do
begin
    if reason = PeerClosed then
        WriteLn("  the feed ended with the peer's reason after %lf", total);

    if reason = Corrupt then
        WriteLn("  a corrupt frame was refused with nothing read");
end;

The program feeds itself three times — failing, clean, corrupt — and prints exactly this:

The network stream: stream of Quote fails NetError over a Connection
  the feed ended with the peer's reason after 9.000000
  the clean feed summed 9.000000
  a corrupt frame was refused with nothing read

The failing feed’s values all arrive — the sum reaches 9.0 — then the reason is delivered: buffered values drain before the failure, delivery order rather than wall-clock order.

The walls

A fallible stream is a spelled type, so the compiler holds both directions of the bargain. Bind a plain generator to a fallible variable — or two different domains together — and the bind refuses:

the generator 'Plain' answers 'stream of int64', which cannot bind the
stream variable 'fallible' declared over 'stream of int64 fails WalkError':
the fails domain is part of the stream type's identity, so the two sides
must agree on it exactly

Walk a fallible stream without consuming its ending, and the loop refuses:

the stream variable 'fallible' can end with a failure from domain
'WalkError', which this loop leaves unconsumed: put 'on fail leave',
'on fail continue', or 'on fail <name> do <statement>' behind the loop body

And an on fail over a stream that cannot fail is refused as the meaningless form it is:

the stream variable 'plain' carries no fails domain, so the loop's
'on fail' form has nothing to consume: a plain stream ends only by
exhaustion

The same discipline the failure channel applies to calls, applied to streams: a failure is consumed where it lands, or the program does not compile.

What composes

The stream adapters are generic over the failure domain — Take, Skip and Chain accept stream of T fails E and answer it, passing a failure through unchanged — and a plain stream instantiates the same adapters with no domain at all. One exported name serves both worlds, which is the generics story doing exactly what it was built to do.