Creates a new Subscription instance, the fundamental datatype and building block provided by this package.
remarks
Most other types in rstream, including Streams, are Subscriptions and all can be:
connected into directed graphs (sync or async & not necessarily DAGs)
transformed using transducers (incl. support for early termination)
can have any number of subscribers (optionally each w/ their own transducers)
recursively unsubscribe themselves from parent after their last subscriber unsubscribed (configurable)
will go into a non-recoverable error state if none of the subscribers has an error handler itself
implement the {@link @thi.ng/api#IDeref} interface
If a transducer is provided (via the xform option), all received values will be first processed by the transducer and only its transformed result(s) (if any) will be passed to downstream subscribers. Any uncaught errors inside the transducer will cause this subscription's error handler to be called and will stop this subscription from receiving any further values (by default, unless overridden).
Subscription behavior can be customized via the additional (optional) options arg. See CommonOpts and SubscriptionOpts for further details.
example
// as reactive value mechanism (same as with stream() above)
s = subscription();
s.subscribe(trace("s1"));
s.subscribe(trace("s2"), { xform: tx.filter((x) => x > 25) });
// external trigger
s.next(23);
// s1 23
s.next(42);
// s1 42// s2 42
Creates a new Subscription instance, the fundamental datatype and building block provided by this package.
Most other types in rstream, including Streams, are
Subscriptions and all can be:If a transducer is provided (via the
xformoption), all received values will be first processed by the transducer and only its transformed result(s) (if any) will be passed to downstream subscribers. Any uncaught errors inside the transducer will cause this subscription's error handler to be called and will stop this subscription from receiving any further values (by default, unless overridden).Subscription behavior can be customized via the additional (optional) options arg. See CommonOpts and SubscriptionOpts for further details.
// as reactive value mechanism (same as with stream() above) s = subscription(); s.subscribe(trace("s1")); s.subscribe(trace("s2"), { xform: tx.filter((x) => x > 25) }); // external trigger s.next(23); // s1 23 s.next(42); // s1 42 // s2 42