stream

A library of channel-backed sources, transforms, and sinks. Streams are simply Channel values used as data flows — the underlying primitive is unchanged. Each source and each transform is a stage: a task that reads its input channel, calls the user closure, and writes to the output channel. A stage waits like any task, so a pipeline of many stages holds no thread while it waits. Backpressure is provided by channel capacity (default 16; configurable via stream.buffered).

Sinks (collect, fold, count, etc.) drain a channel in the calling task, which waits for each value.

See also io / fs for the underlying file operations behind file_chunks / file_lines, tcp for tcp_chunks / tcp_lines, and channel / task for the primitive channel operations.

Summary

Sources

Transforms

Combinators

Sinks

Examples

Three-step pipeline

import stream

fn main() {
  let squares = stream.from_range(1, 100)
    |> stream.filter { n -> n % 2 == 1 }
    |> stream.map { n -> n * n }
    |> stream.take(5)
    |> stream.collect
  println(squares)
}

Generator via unfold

import stream

fn main() {
  -- Generate 1, 2, 3, 4, 5 then None.
  let xs = stream.collect(
    stream.unfold(1) { n ->
      match n > 5 {
        true -> None
        false -> Some((n, n + 1))
      }
    },
  )
  println(xs)
}

Design notes

• Streams are channels. No new value type. stream.collect(ch) works on any Channel, not just streams produced by this module.
• Backpressure is automatic. When the output channel of a stage is full, the stage waits for room, and so consumes no further input.
• A failing stage fails the pipeline. If the function of a stage raises a runtime error (a division by zero, a panic), the stage closes its output with that failure. The next stage fails with it in turn, and the sink at the end raises it: stream.collect does not return the values that came before the failure. channel.each on a stream raises it too; a plain channel.receive sees Closed.
• A pipeline that is cut short stops. When a stage or a sink reads no further (take, take_while, first, a failure downstream), the stage that feeds it is stopped, and so on up to the source. So stream.repeat(x) |> stream.take(3) leaves nothing running. The output of a stage is closed by this: read it with one consumer. A channel that no stage feeds (one you made with channel.new) is left open. What is left in it is another matter: a stage reads ahead of what its reader asks for (a map stage until its output is full, a filter stage as far as it finds nothing to pass on), and what a stage took is gone when the pipeline is cut short. stream.take directly on a channel takes exactly what it gives.
• A TCP source or sink that is stopped closes its connection. When stream.tcp_lines or stream.tcp_chunks is stopped before the peer has closed (stream.tcp_lines(conn) |> stream.first), or the task in stream.write_to_tcp is cancelled, the connection is shut down, as by tcp.close: a read that may be in flight has to end, and what the source had read ahead is gone anyway. That holds at whatever moment the source is stopped once it has begun to read: in a read, or with an item that nobody takes any more. (A source that is stopped before it has read anything leaves the connection as it is.) Read with tcp.read if the connection is to be used afterwards.
• A stream that you stop reading goes on. A stream that is read by hand (channel.receive, channel.select) and then left is still at work: its stages run until their outputs are full or the source is at its end, and the program does not end before they have stopped (see When a program ends). channel.close(s) on the stream stops them at once.
• Errors flow through the stream. File sources emit Channel(Result(_, IoError)); TCP sources emit Channel(Result(_, TcpError)). Each chunk can fail independently; consumers pattern-match. Use map_ok / filter_ok to apply transformations only to Ok values, passing Err(_) through unchanged.
• No async/await. Stages are tasks of the silt scheduler; file and socket reads run on the I/O pool while the stage waits.
• stream.repeat is infinite. Always pair it with take, take_while, or another bounded sink — collect on an unbounded stream will hang.

Forward compatibility

Function names mirror what method-form dispatch (s.map(f)) would look like once silt grows a Stream trait. Existing silt programs will continue to compile and behave identically when that trait lands.