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 examples/NetValueStream run
make -C 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;
    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 that 7.0.0 introduces — 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 onto the instance’s own stack 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'

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

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

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.