123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120 |
- /** @prettier */
- import { Subscription } from './Subscription';
- import { Subscriber } from './Subscriber';
- import { Observable } from './Observable';
- import { subscribeTo } from './util/subscribeTo';
- interface SimpleOuterSubscriberLike<T> {
- /**
- * A handler for inner next notifications from the inner subscription
- * @param innerValue the value nexted by the inner producer
- */
- notifyNext(innerValue: T): void;
- /**
- * A handler for inner error notifications from the inner subscription
- * @param err the error from the inner producer
- */
- notifyError(err: any): void;
- /**
- * A handler for inner complete notifications from the inner subscription.
- */
- notifyComplete(): void;
- }
- export class SimpleInnerSubscriber<T> extends Subscriber<T> {
- constructor(private parent: SimpleOuterSubscriberLike<any>) {
- super();
- }
- protected _next(value: T): void {
- this.parent.notifyNext(value);
- }
- protected _error(error: any): void {
- this.parent.notifyError(error);
- this.unsubscribe();
- }
- protected _complete(): void {
- this.parent.notifyComplete();
- this.unsubscribe();
- }
- }
- export class ComplexInnerSubscriber<T, R> extends Subscriber<R> {
- constructor(private parent: ComplexOuterSubscriber<T, R>, public outerValue: T, public outerIndex: number) {
- super();
- }
- protected _next(value: R): void {
- this.parent.notifyNext(this.outerValue, value, this.outerIndex, this);
- }
- protected _error(error: any): void {
- this.parent.notifyError(error);
- this.unsubscribe();
- }
- protected _complete(): void {
- this.parent.notifyComplete(this);
- this.unsubscribe();
- }
- }
- export class SimpleOuterSubscriber<T, R> extends Subscriber<T> implements SimpleOuterSubscriberLike<R> {
- notifyNext(innerValue: R): void {
- this.destination.next(innerValue);
- }
- notifyError(err: any): void {
- this.destination.error(err);
- }
- notifyComplete(): void {
- this.destination.complete();
- }
- }
- /**
- * DO NOT USE (formerly "OuterSubscriber")
- * TODO: We want to refactor this and remove it. It is retaining values it shouldn't for long
- * periods of time.
- */
- export class ComplexOuterSubscriber<T, R> extends Subscriber<T> {
- /**
- * @param _outerValue Used by: bufferToggle, delayWhen, windowToggle
- * @param innerValue Used by: subclass default, combineLatest, race, bufferToggle, windowToggle, withLatestFrom
- * @param _outerIndex Used by: combineLatest, race, withLatestFrom
- * @param _innerSub Used by: delayWhen
- */
- notifyNext(_outerValue: T, innerValue: R, _outerIndex: number, _innerSub: ComplexInnerSubscriber<T, R>): void {
- this.destination.next(innerValue);
- }
- notifyError(error: any): void {
- this.destination.error(error);
- }
- /**
- * @param _innerSub Used by: race, bufferToggle, delayWhen, windowToggle, windowWhen
- */
- notifyComplete(_innerSub: ComplexInnerSubscriber<T, R>): void {
- this.destination.complete();
- }
- }
- export function innerSubscribe(result: any, innerSubscriber: Subscriber<any>): Subscription | undefined {
- if (innerSubscriber.closed) {
- return undefined;
- }
- if (result instanceof Observable) {
- return result.subscribe(innerSubscriber);
- }
- let subscription: Subscription;
- try {
- subscription = subscribeTo(result)(innerSubscriber) as Subscription;
- } catch (error) {
- innerSubscriber.error(error);
- }
- return subscription;
- }
|