Module "stream-merge"

Const merge

• merge<A, B>(opts?: Partial<StreamMergeOpts<A, B>>): StreamMerge<A, B>

• Returns a new StreamMerge subscription, consuming values from multiple inputs and passing received values on to any subscribers.

  remarks

  Input streams can be added and removed dynamically. By default, StreamMerge calls ISubscriber.done when the last active input is done, but this behavior can be overridden via the provided options.

  example

  merge({
      // input streams w/ different frequencies
      src: [
          fromIterable([1, 2, 3], { delay: 10 }),
          fromIterable([10, 20, 30], { delay: 21 }),
          fromIterable([100, 200, 300], { delay: 7 })
      ]
  }).subscribe(trace());
  // 100
  // 1
  // 200
  // 10
  // 2
  // 300
  // 3
  // 20
  // 30

  example

  Use the {@link @thi.ng/transducers#(labeled:1)} transducer for each input to create a stream of labeled values and track their provenance:

  example

  merge({
      src: [
          fromIterable([1, 2, 3]).transform(tx.labeled("a")),
          fromIterable([10, 20, 30]).transform(tx.labeled("b")),
      ]
  }).subscribe(trace());
  // ["a", 1]
  // ["b", 10]
  // ["a", 2]
  // ["b", 20]
  // ["a", 3]
  // ["b", 30]

  Type parameters

  • A

  • B

  Parameters

  • Optional opts: Partial<StreamMergeOpts<A, B>>

    -

  Returns StreamMerge<A, B>