Par

Module Stream

Finite pull-based streams with fallible production and infallible close.

Consumers request one item at a time with .next. A requested step can either produce an item, finish successfully, or finish with an error. Consumers may stop early with .close, allowing the producer to release its resources.

type Stream<e, a> = recursive choice {
  .close* => !,
  .next => either {
    .end Try<e, !>,
    .item(a) self,
  },
}

A finite pull-based sequence of values of type a whose production can fail with an error of type e.

Operations:

  • .close abandons the stream and releases its resources.
  • .next requests the next step. It returns .item(value) stream when an item is available, .end.ok! on normal completion, or .end.err e when production fails.
type Stream.Event<e, a> = either {
  .cancelled!,
  .ended Try<e, !>,
  .produced a,
  .spawned!,
}

A lifecycle or production event from a registered stream.

  • .spawned! — the stream was accepted.
  • .produced value — the stream produced a value.
  • .ended result — the stream ended normally or with an error.
  • .cancelled! — the registry closed the stream after cancellation.

Cancellation cannot interrupt an outstanding pull. It may therefore be preceded by one final .produced value, or natural .ended result may win instead.

type Stream.Events<id, e, a> = recursive either {
  .end!,
  .event(id, Stream.Event<e, a>) choice {
    .next* => self,
  },
}

A finite sequence of lifecycle and production events from a stream group.

Events from different streams are interleaved by readiness. For each ID, .spawned! comes first, produced values remain in stream order, and .ended result or .cancelled! terminates its lifecycle before that ID can be reused.

After each event, the consumer must select .next. This acknowledges event delivery, but streams are pulled eagerly: the first pull starts when a stream is spawned, and a subsequent pull may start before the current event is acknowledged. At most one pull is outstanding per stream.

Events may be dropped via auto-cleanup (if the items and errors are droppable), but keep in mind that dropping does not cancel the backing streams in a group or registry. The events will still be generated and consumed in the background.

type Stream.Group<e, a> = iterative choice {
  .cancelAll => self,
  .end => ?,
  .spawn(Stream<e, a>) => self,
}

Controls a dynamic group of concurrently pulled streams.

  • .spawn(stream) adds a stream.
  • .cancelAll requests cooperative cancellation of all currently active streams.
  • .end stops the controller without cancelling those already active.

When served by Group, streams receive consecutive Nat IDs starting at zero.

type Stream.Registry<id, e, a> = iterative choice {
  .alloc(id) => either {
    .free choice {
      .spawn(Stream<e, a>) => self,
    },
    .taken self,
  },
  .cancel(id) => (Bool) self,
  .cancelAll => self,
  .end => ?,
}

Controls a dynamic registry of concurrently pulled streams identified by id.

  • .alloc(id) — tries to reserves an ID for a stream, with two possible results:
    • .free — obliges the controller to provide its stream with .spawn, while
    • .taken — means the ID is already in use, and the stream cannot be spawned.
  • .cancel(id) — requests cooperative cancellation of the stream with the given ID and returns .true! when such stream is currently active. The ID remains taken until its stream cooperatively completes the cancellation.
  • .cancelAll — requests cooperative cancellation of all currently active streams.
  • .end — stops the controller without cancelling active streams.
dec Stream.All : [<e, a> Stream<e, a>, box [a] Bool] Try<e, Bool>

Returns .ok.true! if the test function holds for every item. Short-circuits and closes the source on the first .false!.

dec Stream.Any : [<e, a> Stream<e, a>, box [a] Bool] Try<e, Bool>

Returns .ok.true! if the test function holds for at least one item. Short-circuits and closes the source on the first .true!.

dec Stream.Collect : [<e, a> Stream<e, a>] (Try<e, !>) List<a>

Collects the stream into a list. If the stream fails, the items collected so far are returned along with the error.

dec Stream.Concat : [<e, a> List<Stream<e, a>>] Stream<e, a>

Concatenates a list of streams into a single stream.

dec Stream.Drop : [<e, a: drop> Stream<e, a>, Nat] Stream<e, a>

Skips the requested number of items and yields the rest.

dec Stream.DropWhile : [<e, a: share> Stream<e, a>, box [a] Bool] Stream<e, a>

Skips items while the test function returns .true!, then yields the rest.

dec Stream.Filter : [<e, a: share> Stream<e, a>, box [a] Bool] Stream<e, a>

Keeps the items for which the test function returns .true!.

dec Stream.FlatMap : [<e, a> Stream<e, a>, <b> box [a] Stream<e, b>] Stream<e, b>

Maps each item to a stream, then flattens the result.

dec Stream.ForEach : [<r> r, <e, a> Stream<e, a>, box [r, a] r] (Try<e, !>) r

Folds over the stream. The returned pair contains the stream completion status and the final accumulator.

dec Stream.FromList : [<a: drop> List<a>] Stream<either {}, a>

Creates an infallible stream from a list.

dec Stream.Group : [<e, a> dual Stream.Group<e, a>] Stream.Events<Nat, e, a>

Runs a stream group and returns its readiness-ordered events. The event sequence ends after the controller and all spawned streams have terminated.

let events = Stream.Group(chan group {
  group.spawn(firstStream)
  group.spawn(secondStream)
  group.end!
})

dec Stream.Map : [<e, a> Stream<e, a>, <b> box [a] b] Stream<e, b>

Applies a mapping function to each item of a stream.

dec Stream.MapErr : [<e1, a> Stream<e1, a>, <e2> box [e1] e2] Stream<e2, a>

Applies a mapping function to the stream's error type.

dec Stream.Registry : [<id: data, e, a> dual Stream.Registry<id, e, a>] Stream.Events<id, e, a>

Runs a keyed stream registry and returns its readiness-ordered events. The event sequence ends after the controller and all registered streams have terminated.

let events = Stream.Registry(chan registry {
  registry.alloc(id).case {
    .taken => {}
    .free => { registry.spawn(stream) }
  }
  registry.end!
})

dec Stream.Sum : [<e, a: number> Stream<e, a>] Try<e, a>

Calculates the sum of all stream items.

dec Stream.Take : [<e, a> Stream<e, a>, Nat] Stream<e, a>

Yields at most the requested number of items, closing the source when enough items have been taken.

dec Stream.TakeWhile : [<e, a: share> Stream<e, a>, box [a] Bool] Stream<e, a>

Yields items while the test function returns .true!, then closes the source.

dec Stream.ToList : [<e, a: drop> Stream<e, a>] Try<e, List<a>>

Collects the stream into a list. If the stream fails, the list is discarded and the error is returned.