Module async/stream

Async iteration — the Stream trait: "a value, later, repeatedly" (plans/reference/ASYNC_ITERATION_STREAM.md).

Iterator yields values now; Future yields one value later. A Stream is both: each next(io) returns a FUTURE of Option(Item), and .None means the stream is finished. Before this trait existed, every std API that produced a sequence asynchronously invented its own shape — TcpListener.accept in the caller's own loop, Watcher.next answering Option(FsEvent), Channel.recv plus a separate try_recv — and none of them composed.

{ Stream } :: import("std/async/stream");
{ watch, WatchOptions } :: import("std/fs/watch");

w := watch(Path.new(`./logs`), WatchOptions.defaults(), exn);
names := w.map(e => e.name).take(usize(3));
got := io.await(names.collect(io), io); // ArrayList(String), 3 entries

The combinators (map, filter, filter_map, take, skip) are LAZY: each wraps the upstream stream and pulls one item per next. The consumers (for_each, collect) drive the stream to .None.

Looping over a stream

The loop of async iteration is written out: while + io.await(s.next(io), io)

• match, and its body is ordinary loop code — break, continue and return work in it:

io.async((io : Io) => {
  mut(s) := conns;
  while(true, {
    n := io.await(s.next(io), io);
    match(
      n,
      .Some(c) =>
        match(
          c,
          .Ok(st) => serve(st, io),
          .Err(_) => {
            break;
          }
        ),
      .None => {
        break;
      }
    );
  });
})

The io.await is written in the io.async body, so it suspends the task. A suspension point is an io.await written in the io.async body; an io.await inside a function the task calls blocks instead of suspending, which is why std has no named loop form that hides the await (plans/reference/ASYNC_ITERATION_STREAM.md). for_each(f, io) stays for the combinator style.

Implementing it

next takes self : &mut Self, the way Iterator does, and its future borrows the stream (plans/VALUES_BY_DEFAULT.md §3.13 A2/A5): the body names the borrow in its capture list, and the future is second-class, so it is awaited where it is made (io.await(s.next(io), io)) or returned from a function whose &mut parameter it borrows. A stream SOURCE is therefore a plain value whose fields next writes through the borrow:

Countdown :: struct(_n : i32);
impl(
  Countdown,
  Stream(
    Item : i32,
    next : (fn(self : &mut Self, io : Io) -> Impl(Future(Option(i32), Io)))(
      io.async({ &mut self }(io : Io) =>
        cond(
          (self._n <= i32(0)) => Option(i32).None,
          true => {
            v := self._n;
            self._n = (self._n - i32(1));
            Option(i32).Some(v)
          }
        ))
    )
  )
);

The combinators take the stream and their closure BY VALUE: each adapter (StreamMap, StreamFilter, StreamFilterMap, StreamTake, StreamSkip) is a plain value that owns what it wraps. The consumers (for_each, collect) consume the stream too, so their future owns it and a task may own that future (io.spawn(chain.collect(io), io)).

A stream whose items can FAIL carries the failure in the item — Item = Result(T, E), the way TcpListener.incoming yields Result(TcpStream, IoError) — so one bad item does not end the stream and no consumer needs an Exception handler. That is why next's future is Future(_, Io) and not Future(_, IoExn): a stream never throws.

Writing a chain

Build the chain OUTSIDE the io.async body that awaits it. An => closure passed to a generic callback parameter INSIDE an async body leaves the enclosing future's result type unresolved (issues/fixed/closure-argument-inside-an-io-async-body-loses-the-future-result-type.md), so s.map(f) written inside a task body fails at the task's await. The chain is a value: make it first, then await it wherever you like — including from inside a spawned task.

Stability

unstable — new in v0.2.27; the API may still change until the next release. Two things are most likely to move: the combinator SET (flat_map is deliberately absent — its doubly-derived item type is the shape a var-bound combinator receiver still resolves wrongly).