ꦫꦣꦺꦤ꧀ꦄꦤ꧀ꦠꦱꦺꦤ

PATH: usr / local / lib / node_modules / ghost-cli / node_modules / rxjs / src / internal / operators

import { Operator } from '../Operator';
import { Subscriber } from '../Subscriber';
import { Subscription } from '../Subscription';
import { MonoTypeOperatorFunction, TeardownLogic } from '../types';
import { ConnectableObservable } from '../observable/ConnectableObservable';
import { Observable } from '../Observable';

/**
 * Make a {@link ConnectableObservable} behave like a ordinary observable and automates the way
 * you can connect to it.
 *
 * Internally it counts the subscriptions to the observable and subscribes (only once) to the source if
 * the number of subscriptions is larger than 0. If the number of subscriptions is smaller than 1, it
 * unsubscribes from the source. This way you can make sure that everything before the *published*
 * refCount has only a single subscription independently of the number of subscribers to the target
 * observable.
 *
 * Note that using the {@link share} operator is exactly the same as using the *publish* operator
 * (making the observable hot) and the *refCount* operator in a sequence.
 *
 * ![](refCount.png)
 *
 * ## Example
 *
 * In the following example there are two intervals turned into connectable observables
 * by using the *publish* operator. The first one uses the *refCount* operator, the
 * second one does not use it. You will notice that a connectable observable does nothing
 * until you call its connect function.
 *
 * ```ts
 * import { interval } from 'rxjs';
 * import { tap, publish, refCount } from 'rxjs/operators';
 *
 * // Turn the interval observable into a ConnectableObservable (hot)
 * const refCountInterval = interval(400).pipe(
 *   tap((num) => console.log(`refCount ${num}`)),
 *   publish(),
 *   refCount()
 * );
 *
 * const publishedInterval = interval(400).pipe(
 *   tap((num) => console.log(`publish ${num}`)),
 *   publish()
 * );
 *
 * refCountInterval.subscribe();
 * refCountInterval.subscribe();
 * // 'refCount 0' -----> 'refCount 1' -----> etc
 * // All subscriptions will receive the same value and the tap (and
 * // every other operator) before the publish operator will be executed
 * // only once per event independently of the number of subscriptions.
 *
 * publishedInterval.subscribe();
 * // Nothing happens until you call .connect() on the observable.
 * ```
 *
 * @see {@link ConnectableObservable}
 * @see {@link share}
 * @see {@link publish}
 */
export function refCount<T>(): MonoTypeOperatorFunction<T> {
  return function refCountOperatorFunction(source: ConnectableObservable<T>): Observable<T> {
    return source.lift(new RefCountOperator(source));
  } as MonoTypeOperatorFunction<T>;
}

class RefCountOperator<T> implements Operator<T, T> {
  constructor(private connectable: ConnectableObservable<T>) {
  }
  call(subscriber: Subscriber<T>, source: any): TeardownLogic {

const { connectable } = this;
    (<any> connectable)._refCount++;

const refCounter = new RefCountSubscriber(subscriber, connectable);
    const subscription = source.subscribe(refCounter);

if (!refCounter.closed) {
      (<any> refCounter).connection = connectable.connect();
    }

return subscription;
  }
}

class RefCountSubscriber<T> extends Subscriber<T> {

private connection: Subscription;

constructor(destination: Subscriber<T>,
              private connectable: ConnectableObservable<T>) {
    super(destination);
  }

protected _unsubscribe() {

const { connectable } = this;
    if (!connectable) {
      this.connection = null;
      return;
    }

this.connectable = null;
    const refCount = (<any> connectable)._refCount;
    if (refCount <= 0) {
      this.connection = null;
      return;
    }

(<any> connectable)._refCount = refCount - 1;
    if (refCount > 1) {
      this.connection = null;
      return;
    }

///
    // Compare the local RefCountSubscriber's connection Subscription to the
    // connection Subscription on the shared ConnectableObservable. In cases
    // where the ConnectableObservable source synchronously emits values, and
    // the RefCountSubscriber's downstream Observers synchronously unsubscribe,
    // execution continues to here before the RefCountOperator has a chance to
    // supply the RefCountSubscriber with the shared connection Subscription.
    // For example:
    // ```
    // range(0, 10).pipe(
    //   publish(),
    //   refCount(),
    //   take(5),
    // )
    // .subscribe();
    // ```
    // In order to account for this case, RefCountSubscriber should only dispose
    // the ConnectableObservable's shared connection Subscription if the
    // connection Subscription exists, *and* either:
    //   a. RefCountSubscriber doesn't have a reference to the shared connection
    //      Subscription yet, or,
    //   b. RefCountSubscriber's connection Subscription reference is identical
    //      to the shared connection Subscription
    ///
    const { connection } = this;
    const sharedConnection = (<any> connectable)._connection;
    this.connection = null;

if (sharedConnection && (!connection || sharedConnection === connection)) {
      sharedConnection.unsubscribe();
    }
  }
}

[-] toArray.ts [edit]
[-] window.ts [edit]
[-] retry.ts [edit]
[-] mergeAll.ts [edit]
[-] windowWhen.ts [edit]
[-] every.ts [edit]
[-] timeout.ts [edit]
[-] debounceTime.ts [edit]
[-] index.ts [edit]
[-] takeWhile.ts [edit]
[-] audit.ts [edit]
[-] reduce.ts [edit]
[-] mapTo.ts [edit]
[-] count.ts [edit]
[-] debounce.ts [edit]
[-] throwIfEmpty.ts [edit]
[-] auditTime.ts [edit]
[-] scan.ts [edit]
[-] switchAll.ts [edit]
[-] map.ts [edit]
[-] bufferCount.ts [edit]
[-] mergeScan.ts [edit]
[-] publish.ts [edit]
[-] min.ts [edit]
[+] ..
[-] publishReplay.ts [edit]
[-] bufferTime.ts [edit]
[-] share.ts [edit]
[-] repeat.ts [edit]
[-] distinct.ts [edit]
[-] concatMapTo.ts [edit]
[-] buffer.ts [edit]
[-] concatMap.ts [edit]
[-] max.ts [edit]
[-] partition.ts [edit]
[-] observeOn.ts [edit]
[-] bufferWhen.ts [edit]
[-] dematerialize.ts [edit]
[-] throttle.ts [edit]
[-] groupBy.ts [edit]
[-] withLatestFrom.ts [edit]
[-] exhaust.ts [edit]
[-] exhaustMap.ts [edit]
[-] retryWhen.ts [edit]
[-] zipAll.ts [edit]
[-] find.ts [edit]
[-] mergeMap.ts [edit]
[-] single.ts [edit]
[-] last.ts [edit]
[-] combineLatest.ts [edit]
[-] publishLast.ts [edit]
[-] pluck.ts [edit]
[-] pairwise.ts [edit]
[-] distinctUntilChanged.ts [edit]
[-] timeInterval.ts [edit]
[-] shareReplay.ts [edit]
[-] skipUntil.ts [edit]
[-] findIndex.ts [edit]
[-] windowToggle.ts [edit]
[-] startWith.ts [edit]
[-] timeoutWith.ts [edit]
[-] concat.ts [edit]
[-] materialize.ts [edit]
[-] timestamp.ts [edit]
[-] onErrorResumeNext.ts [edit]
[-] subscribeOn.ts [edit]
[-] takeUntil.ts [edit]
[-] skipLast.ts [edit]
[-] throttleTime.ts [edit]
[-] mergeMapTo.ts [edit]
[-] multicast.ts [edit]
[-] merge.ts [edit]
[-] takeLast.ts [edit]
[-] delayWhen.ts [edit]
[-] sampleTime.ts [edit]
[-] tap.ts [edit]
[-] take.ts [edit]
[-] sample.ts [edit]
[-] filter.ts [edit]
[-] skip.ts [edit]
[-] expand.ts [edit]
[-] publishBehavior.ts [edit]
[-] sequenceEqual.ts [edit]
[-] ignoreElements.ts [edit]
[-] catchError.ts [edit]
[-] endWith.ts [edit]
[-] refCount.ts [edit]
[-] windowCount.ts [edit]
[-] defaultIfEmpty.ts [edit]
[-] windowTime.ts [edit]
[-] switchMapTo.ts [edit]
[-] switchMap.ts [edit]
[-] delay.ts [edit]
[-] bufferToggle.ts [edit]
[-] repeatWhen.ts [edit]
[-] concatAll.ts [edit]
[-] first.ts [edit]
[-] zip.ts [edit]
[-] elementAt.ts [edit]
[-] skipWhile.ts [edit]
[-] race.ts [edit]
[-] finalize.ts [edit]
[-] combineAll.ts [edit]
[-] distinctUntilKeyChanged.ts [edit]
[-] isEmpty.ts [edit]