diff options
Diffstat (limited to 'node_modules/rxjs/src/internal/operators/share.ts')
| -rw-r--r-- | node_modules/rxjs/src/internal/operators/share.ts | 267 |
1 files changed, 267 insertions, 0 deletions
diff --git a/node_modules/rxjs/src/internal/operators/share.ts b/node_modules/rxjs/src/internal/operators/share.ts new file mode 100644 index 0000000..bc0c270 --- /dev/null +++ b/node_modules/rxjs/src/internal/operators/share.ts @@ -0,0 +1,267 @@ +import { innerFrom } from '../observable/innerFrom'; +import { Subject } from '../Subject'; +import { SafeSubscriber } from '../Subscriber'; +import { Subscription } from '../Subscription'; +import { MonoTypeOperatorFunction, SubjectLike, ObservableInput } from '../types'; +import { operate } from '../util/lift'; + +export interface ShareConfig<T> { + /** + * The factory used to create the subject that will connect the source observable to + * multicast consumers. + */ + connector?: () => SubjectLike<T>; + /** + * If `true`, the resulting observable will reset internal state on error from source and return to a "cold" state. This + * allows the resulting observable to be "retried" in the event of an error. + * If `false`, when an error comes from the source it will push the error into the connecting subject, and the subject + * will remain the connecting subject, meaning the resulting observable will not go "cold" again, and subsequent retries + * or resubscriptions will resubscribe to that same subject. In all cases, RxJS subjects will emit the same error again, however + * {@link ReplaySubject} will also push its buffered values before pushing the error. + * It is also possible to pass a notifier factory returning an `ObservableInput` instead which grants more fine-grained + * control over how and when the reset should happen. This allows behaviors like conditional or delayed resets. + */ + resetOnError?: boolean | ((error: any) => ObservableInput<any>); + /** + * If `true`, the resulting observable will reset internal state on completion from source and return to a "cold" state. This + * allows the resulting observable to be "repeated" after it is done. + * If `false`, when the source completes, it will push the completion through the connecting subject, and the subject + * will remain the connecting subject, meaning the resulting observable will not go "cold" again, and subsequent repeats + * or resubscriptions will resubscribe to that same subject. + * It is also possible to pass a notifier factory returning an `ObservableInput` instead which grants more fine-grained + * control over how and when the reset should happen. This allows behaviors like conditional or delayed resets. + */ + resetOnComplete?: boolean | (() => ObservableInput<any>); + /** + * If `true`, when the number of subscribers to the resulting observable reaches zero due to those subscribers unsubscribing, the + * internal state will be reset and the resulting observable will return to a "cold" state. This means that the next + * time the resulting observable is subscribed to, a new subject will be created and the source will be subscribed to + * again. + * If `false`, when the number of subscribers to the resulting observable reaches zero due to unsubscription, the subject + * will remain connected to the source, and new subscriptions to the result will be connected through that same subject. + * It is also possible to pass a notifier factory returning an `ObservableInput` instead which grants more fine-grained + * control over how and when the reset should happen. This allows behaviors like conditional or delayed resets. + */ + resetOnRefCountZero?: boolean | (() => ObservableInput<any>); +} + +export function share<T>(): MonoTypeOperatorFunction<T>; + +export function share<T>(options: ShareConfig<T>): MonoTypeOperatorFunction<T>; + +/** + * Returns a new Observable that multicasts (shares) the original Observable. As long as there is at least one + * Subscriber this Observable will be subscribed and emitting data. When all subscribers have unsubscribed it will + * unsubscribe from the source Observable. Because the Observable is multicasting it makes the stream `hot`. + * This is an alias for `multicast(() => new Subject()), refCount()`. + * + * The subscription to the underlying source Observable can be reset (unsubscribe and resubscribe for new subscribers), + * if the subscriber count to the shared observable drops to 0, or if the source Observable errors or completes. It is + * possible to use notifier factories for the resets to allow for behaviors like conditional or delayed resets. Please + * note that resetting on error or complete of the source Observable does not behave like a transparent retry or restart + * of the source because the error or complete will be forwarded to all subscribers and their subscription will be + * closed. Only new subscribers after a reset on error or complete happened will cause a fresh subscription to the + * source. To achieve transparent retries or restarts pipe the source through appropriate operators before sharing. + * + *  + * + * ## Example + * + * Generate new multicast Observable from the `source` Observable value + * + * ```ts + * import { interval, tap, map, take, share } from 'rxjs'; + * + * const source = interval(1000).pipe( + * tap(x => console.log('Processing: ', x)), + * map(x => x * x), + * take(6), + * share() + * ); + * + * source.subscribe(x => console.log('subscription 1: ', x)); + * source.subscribe(x => console.log('subscription 2: ', x)); + * + * // Logs: + * // Processing: 0 + * // subscription 1: 0 + * // subscription 2: 0 + * // Processing: 1 + * // subscription 1: 1 + * // subscription 2: 1 + * // Processing: 2 + * // subscription 1: 4 + * // subscription 2: 4 + * // Processing: 3 + * // subscription 1: 9 + * // subscription 2: 9 + * // Processing: 4 + * // subscription 1: 16 + * // subscription 2: 16 + * // Processing: 5 + * // subscription 1: 25 + * // subscription 2: 25 + * ``` + * + * ## Example with notifier factory: Delayed reset + * + * ```ts + * import { interval, take, share, timer } from 'rxjs'; + * + * const source = interval(1000).pipe( + * take(3), + * share({ + * resetOnRefCountZero: () => timer(1000) + * }) + * ); + * + * const subscriptionOne = source.subscribe(x => console.log('subscription 1: ', x)); + * setTimeout(() => subscriptionOne.unsubscribe(), 1300); + * + * setTimeout(() => source.subscribe(x => console.log('subscription 2: ', x)), 1700); + * + * setTimeout(() => source.subscribe(x => console.log('subscription 3: ', x)), 5000); + * + * // Logs: + * // subscription 1: 0 + * // (subscription 1 unsubscribes here) + * // (subscription 2 subscribes here ~400ms later, source was not reset) + * // subscription 2: 1 + * // subscription 2: 2 + * // (subscription 2 unsubscribes here) + * // (subscription 3 subscribes here ~2000ms later, source did reset before) + * // subscription 3: 0 + * // subscription 3: 1 + * // subscription 3: 2 + * ``` + * + * @see {@link shareReplay} + * + * @return A function that returns an Observable that mirrors the source. + */ +export function share<T>(options: ShareConfig<T> = {}): MonoTypeOperatorFunction<T> { + const { connector = () => new Subject<T>(), resetOnError = true, resetOnComplete = true, resetOnRefCountZero = true } = options; + // It's necessary to use a wrapper here, as the _operator_ must be + // referentially transparent. Otherwise, it cannot be used in calls to the + // static `pipe` function - to create a partial pipeline. + // + // The _operator function_ - the function returned by the _operator_ - will + // not be referentially transparent - as it shares its source - but the + // _operator function_ is called when the complete pipeline is composed via a + // call to a source observable's `pipe` method - not when the static `pipe` + // function is called. + return (wrapperSource) => { + let connection: SafeSubscriber<T> | undefined; + let resetConnection: Subscription | undefined; + let subject: SubjectLike<T> | undefined; + let refCount = 0; + let hasCompleted = false; + let hasErrored = false; + + const cancelReset = () => { + resetConnection?.unsubscribe(); + resetConnection = undefined; + }; + // Used to reset the internal state to a "cold" + // state, as though it had never been subscribed to. + const reset = () => { + cancelReset(); + connection = subject = undefined; + hasCompleted = hasErrored = false; + }; + const resetAndUnsubscribe = () => { + // We need to capture the connection before + // we reset (if we need to reset). + const conn = connection; + reset(); + conn?.unsubscribe(); + }; + + return operate<T, T>((source, subscriber) => { + refCount++; + if (!hasErrored && !hasCompleted) { + cancelReset(); + } + + // Create the subject if we don't have one yet. Grab a local reference to + // it as well, which avoids non-null assertions when using it and, if we + // connect to it now, then error/complete need a reference after it was + // reset. + const dest = (subject = subject ?? connector()); + + // Add the finalization directly to the subscriber - instead of returning it - + // so that the handling of the subscriber's unsubscription will be wired + // up _before_ the subscription to the source occurs. This is done so that + // the assignment to the source connection's `closed` property will be seen + // by synchronous firehose sources. + subscriber.add(() => { + refCount--; + + // If we're resetting on refCount === 0, and it's 0, we only want to do + // that on "unsubscribe", really. Resetting on error or completion is a different + // configuration. + if (refCount === 0 && !hasErrored && !hasCompleted) { + resetConnection = handleReset(resetAndUnsubscribe, resetOnRefCountZero); + } + }); + + // The following line adds the subscription to the subscriber passed. + // Basically, `subscriber === dest.subscribe(subscriber)` is `true`. + dest.subscribe(subscriber); + + if ( + !connection && + // Check this shareReplay is still activate - it can be reset to 0 + // and be "unsubscribed" _before_ it actually subscribes. + // If we were to subscribe then, it'd leak and get stuck. + refCount > 0 + ) { + // We need to create a subscriber here - rather than pass an observer and + // assign the returned subscription to connection - because it's possible + // for reentrant subscriptions to the shared observable to occur and in + // those situations we want connection to be already-assigned so that we + // don't create another connection to the source. + connection = new SafeSubscriber({ + next: (value) => dest.next(value), + error: (err) => { + hasErrored = true; + cancelReset(); + resetConnection = handleReset(reset, resetOnError, err); + dest.error(err); + }, + complete: () => { + hasCompleted = true; + cancelReset(); + resetConnection = handleReset(reset, resetOnComplete); + dest.complete(); + }, + }); + innerFrom(source).subscribe(connection); + } + })(wrapperSource); + }; +} + +function handleReset<T extends unknown[] = never[]>( + reset: () => void, + on: boolean | ((...args: T) => ObservableInput<any>), + ...args: T +): Subscription | undefined { + if (on === true) { + reset(); + return; + } + + if (on === false) { + return; + } + + const onSubscriber = new SafeSubscriber({ + next: () => { + onSubscriber.unsubscribe(); + reset(); + }, + }); + + return innerFrom(on(...args)).subscribe(onSubscriber); +} |