@metreeca/flow - v0.10.0
    Preparing search index...

    Function join

    Creates a task interleaving nested feeds into a single feed, with or without a task opening the feeds to interleave.

    • Creates a task interleaving a feed of feeds into a single feed.

      Nested feeds are opened as the source yields them and consumed together, each kept one item ahead, so a slow nested feed never holds back the others; the items of every nested feed keep their own order among themselves. A nested feed contributing nothing simply drops out.

      Warning

      • Incremental: items are emitted as the nested feeds report them, so the reported feed runs dry as the source and its nested feeds do, an infinite nested feed keeping it open without holding back the items of the others.
      • Materialising: a pending item is held for every nested feed open at the same time, and nothing bounds their number, so a source yielding feeds faster than they run dry may exhaust memory; splice with flat() instead where the source carries an unbounded number of feeds.
      • Stateless: nested feeds are interleaved without state carried across them.
      Warning

      Output order is not preserved: items interleave and overtake each other according to how quickly every nested feed produces them.

      Note

      Nested feeds failing while the consumer is idle report their error when the feed is next advanced, rather than escaping as unhandled rejections.

      Note

      The source and every nested feed still open are closed when the feed is exhausted, fails or is closed early, waiting for their pending items to settle first, so a feed idling between items delays it; failures reported while closing are suppressed.

      Type Parameters

      • V

        The type of items carried by the nested feeds

      Returns Task<Feed<V>, V>

      A task yielding the items of every nested feed as they become available

      await pipe(
        (items([slow, fast]))  // slow yields 1, 2; fast yields 3, 4
        (join())
        (toArray())
      );  // [3, 4, 1, 2], as the faster feed reports first

    • Creates a task interleaving the feeds another task reports into a single feed.

      Hands the feed drawn from to task and interleaves the feeds it reports as the overload taking no argument does, so that an item mapped to a feed of its own is expanded in place into the items of that feed.

      Warning

      • Incremental: items are emitted as the reported feeds report them, so the reported feed runs dry as the source, task and the feeds it reports do, an infinite one keeping it open without holding back the items of the others.
      • Materialising: a pending item is held for every reported feed open at the same time, and nothing bounds their number, so feeds opened faster than they run dry may exhaust memory; splice with flat() instead where task reports an unbounded number of feeds.
      • Stateless: the interleaving carries no state across the reported feeds, whatever task carries across the items it draws.
      Note

      task draws from the whole feed, so state it initialises on invocation decides on every item, as it would anywhere else in the pipe. Where a source already carries feeds and a task is to be scoped to each of them, apply it within map(): join(map(feed => feed(take(2)))) yields its quota to every nested feed.

      Type Parameters

      • V

        The type of items drawn from the feed

      • R

        The type of items carried by the feeds task reports

      Parameters

      • task: Task<V, Feed<R>>

        The task opening the feeds to interleave

      Returns Task<V, R>

      A task yielding the items of every feed task reports, as they become available

      await pipe(
        (items([30, 10]))  // the delay of each retrieval
        (join(map(ms => retrieve(ms))))
        (toArray())
      );  // the items of the faster retrieval first

    Settings

    Member Visibility