Returns a Subscription which transforms each incoming value into a new Stream, subscribes to it (via an hidden / internal subscription) and then only passes values from that stream to its own subscribers.
remarks
If a new value is received, the metastream first unsubscribes from any still active stream, before creating and subscribing to the new stream. Hence this stream type is useful for cases where streams need to be dynamically created & inserted into an existing dataflow topology.
The user supplied factory function will be called for each incoming value and is responsible for creating the new stream instances. If the function returns null/undefined, no further action will be taken (acts like a filter transducer).
The factory function does NOT need to create new streams, but can merely return other existing streams, and so making the meta stream act like a switch with arbitrary criteria.
If the meta stream itself is the only subscriber to existing input streams, you'll need to configure the input's CommonOpts.closeOut option to keep them alive and support dynamic switching between them.
example
// transform each received odd number into a stream// producing 3 copies of that number in the metastream// even numbers are ignored
a = metastream(
(x) => (x & 1)
? fromIterable(tx.repeat(x, 3), { delay: 100 })
: null
);
a.subscribe(trace())
a.next(23)
// 23// 23// 23
a.next(42) // ignored by factory fn
a.next(43)
// 43// 43// 43
example
// infinite inputs
a = fromIterable(
tx.repeat("a"),
{ delay: 1000, closeOut: CloseMode.NEVER }
);
b = fromIterable(
tx.repeat("b"),
{ delay: 1000, closeOut: CloseMode.NEVER }
);
// stream selector / switch
m = metaStream((x) => x ? a : b);
m.subscribe(trace("meta from: "));
m.next(true);
// meta from: a
m.next(false);
// meta from: b
m.next(true);
// meta from: a
Returns a Subscription which transforms each incoming value into a new Stream, subscribes to it (via an hidden / internal subscription) and then only passes values from that stream to its own subscribers.
If a new value is received, the metastream first unsubscribes from any still active stream, before creating and subscribing to the new stream. Hence this stream type is useful for cases where streams need to be dynamically created & inserted into an existing dataflow topology.
The user supplied
factoryfunction will be called for each incoming value and is responsible for creating the new stream instances. If the function returns null/undefined, no further action will be taken (acts like a filter transducer).The factory function does NOT need to create new streams, but can merely return other existing streams, and so making the meta stream act like a switch with arbitrary criteria.
If the meta stream itself is the only subscriber to existing input streams, you'll need to configure the input's CommonOpts.closeOut option to keep them alive and support dynamic switching between them.
// transform each received odd number into a stream // producing 3 copies of that number in the metastream // even numbers are ignored a = metastream( (x) => (x & 1) ? fromIterable(tx.repeat(x, 3), { delay: 100 }) : null ); a.subscribe(trace()) a.next(23) // 23 // 23 // 23 a.next(42) // ignored by factory fn a.next(43) // 43 // 43 // 43// infinite inputs a = fromIterable( tx.repeat("a"), { delay: 1000, closeOut: CloseMode.NEVER } ); b = fromIterable( tx.repeat("b"), { delay: 1000, closeOut: CloseMode.NEVER } ); // stream selector / switch m = metaStream((x) => x ? a : b); m.subscribe(trace("meta from: ")); m.next(true); // meta from: a m.next(false); // meta from: b m.next(true); // meta from: a