Server IP : 150.95.80.236 / Your IP : 18.117.107.40 Web Server : Apache System : Linux host-150-95-80-236 3.10.0-1160.105.1.el7.x86_64 #1 SMP Thu Dec 7 15:39:45 UTC 2023 x86_64 User : social-telecare ( 10000) PHP Version : 7.4.33 Disable Function : opcache_get_status MySQL : OFF | cURL : ON | WGET : OFF | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /var/www/vhosts/pcu.in.th/api-uat.pcu.in.th/node_modules/rxjs/src/internal/ |
Upload File : |
import { isFunction } from './util/isFunction'; import { Observer, ObservableNotification } from './types'; import { isSubscription, Subscription } from './Subscription'; import { config } from './config'; import { reportUnhandledError } from './util/reportUnhandledError'; import { noop } from './util/noop'; import { nextNotification, errorNotification, COMPLETE_NOTIFICATION } from './NotificationFactories'; import { timeoutProvider } from './scheduler/timeoutProvider'; import { captureError } from './util/errorContext'; /** * Implements the {@link Observer} interface and extends the * {@link Subscription} class. While the {@link Observer} is the public API for * consuming the values of an {@link Observable}, all Observers get converted to * a Subscriber, in order to provide Subscription-like capabilities such as * `unsubscribe`. Subscriber is a common type in RxJS, and crucial for * implementing operators, but it is rarely used as a public API. * * @class Subscriber<T> */ export class Subscriber<T> extends Subscription implements Observer<T> { /** * A static factory for a Subscriber, given a (potentially partial) definition * of an Observer. * @param next The `next` callback of an Observer. * @param error The `error` callback of an * Observer. * @param complete The `complete` callback of an * Observer. * @return A Subscriber wrapping the (partially defined) * Observer represented by the given arguments. * @nocollapse * @deprecated Do not use. Will be removed in v8. There is no replacement for this * method, and there is no reason to be creating instances of `Subscriber` directly. * If you have a specific use case, please file an issue. */ static create<T>(next?: (x?: T) => void, error?: (e?: any) => void, complete?: () => void): Subscriber<T> { return new SafeSubscriber(next, error, complete); } /** @deprecated Internal implementation detail, do not use directly. Will be made internal in v8. */ protected isStopped: boolean = false; /** @deprecated Internal implementation detail, do not use directly. Will be made internal in v8. */ protected destination: Subscriber<any> | Observer<any>; // this `any` is the escape hatch to erase extra type param (e.g. R) /** * @deprecated Internal implementation detail, do not use directly. Will be made internal in v8. * There is no reason to directly create an instance of Subscriber. This type is exported for typings reasons. */ constructor(destination?: Subscriber<any> | Observer<any>) { super(); if (destination) { this.destination = destination; // Automatically chain subscriptions together here. // if destination is a Subscription, then it is a Subscriber. if (isSubscription(destination)) { destination.add(this); } } else { this.destination = EMPTY_OBSERVER; } } /** * The {@link Observer} callback to receive notifications of type `next` from * the Observable, with a value. The Observable may call this method 0 or more * times. * @param {T} [value] The `next` value. * @return {void} */ next(value?: T): void { if (this.isStopped) { handleStoppedNotification(nextNotification(value), this); } else { this._next(value!); } } /** * The {@link Observer} callback to receive notifications of type `error` from * the Observable, with an attached `Error`. Notifies the Observer that * the Observable has experienced an error condition. * @param {any} [err] The `error` exception. * @return {void} */ error(err?: any): void { if (this.isStopped) { handleStoppedNotification(errorNotification(err), this); } else { this.isStopped = true; this._error(err); } } /** * The {@link Observer} callback to receive a valueless notification of type * `complete` from the Observable. Notifies the Observer that the Observable * has finished sending push-based notifications. * @return {void} */ complete(): void { if (this.isStopped) { handleStoppedNotification(COMPLETE_NOTIFICATION, this); } else { this.isStopped = true; this._complete(); } } unsubscribe(): void { if (!this.closed) { this.isStopped = true; super.unsubscribe(); this.destination = null!; } } protected _next(value: T): void { this.destination.next(value); } protected _error(err: any): void { try { this.destination.error(err); } finally { this.unsubscribe(); } } protected _complete(): void { try { this.destination.complete(); } finally { this.unsubscribe(); } } } /** * This bind is captured here because we want to be able to have * compatibility with monoid libraries that tend to use a method named * `bind`. In particular, a library called Monio requires this. */ const _bind = Function.prototype.bind; function bind<Fn extends (...args: any[]) => any>(fn: Fn, thisArg: any): Fn { return _bind.call(fn, thisArg); } /** * Internal optimization only, DO NOT EXPOSE. * @internal */ class ConsumerObserver<T> implements Observer<T> { constructor(private partialObserver: Partial<Observer<T>>) {} next(value: T): void { const { partialObserver } = this; if (partialObserver.next) { try { partialObserver.next(value); } catch (error) { handleUnhandledError(error); } } } error(err: any): void { const { partialObserver } = this; if (partialObserver.error) { try { partialObserver.error(err); } catch (error) { handleUnhandledError(error); } } else { handleUnhandledError(err); } } complete(): void { const { partialObserver } = this; if (partialObserver.complete) { try { partialObserver.complete(); } catch (error) { handleUnhandledError(error); } } } } export class SafeSubscriber<T> extends Subscriber<T> { constructor( observerOrNext?: Partial<Observer<T>> | ((value: T) => void) | null, error?: ((e?: any) => void) | null, complete?: (() => void) | null ) { super(); let partialObserver: Partial<Observer<T>>; if (isFunction(observerOrNext) || !observerOrNext) { // The first argument is a function, not an observer. The next // two arguments *could* be observers, or they could be empty. partialObserver = { next: (observerOrNext ?? undefined) as (((value: T) => void) | undefined), error: error ?? undefined, complete: complete ?? undefined, }; } else { // The first argument is a partial observer. let context: any; if (this && config.useDeprecatedNextContext) { // This is a deprecated path that made `this.unsubscribe()` available in // next handler functions passed to subscribe. This only exists behind a flag // now, as it is *very* slow. context = Object.create(observerOrNext); context.unsubscribe = () => this.unsubscribe(); partialObserver = { next: observerOrNext.next && bind(observerOrNext.next, context), error: observerOrNext.error && bind(observerOrNext.error, context), complete: observerOrNext.complete && bind(observerOrNext.complete, context), }; } else { // The "normal" path. Just use the partial observer directly. partialObserver = observerOrNext; } } // Wrap the partial observer to ensure it's a full observer, and // make sure proper error handling is accounted for. this.destination = new ConsumerObserver(partialObserver); } } function handleUnhandledError(error: any) { if (config.useDeprecatedSynchronousErrorHandling) { captureError(error); } else { // Ideal path, we report this as an unhandled error, // which is thrown on a new call stack. reportUnhandledError(error); } } /** * An error handler used when no error handler was supplied * to the SafeSubscriber -- meaning no error handler was supplied * do the `subscribe` call on our observable. * @param err The error to handle */ function defaultErrorHandler(err: any) { throw err; } /** * A handler for notifications that cannot be sent to a stopped subscriber. * @param notification The notification being sent * @param subscriber The stopped subscriber */ function handleStoppedNotification(notification: ObservableNotification<any>, subscriber: Subscriber<any>) { const { onStoppedNotification } = config; onStoppedNotification && timeoutProvider.setTimeout(() => onStoppedNotification(notification, subscriber)); } /** * The observer used as a stub for subscriptions where the user did not * pass any arguments to `subscribe`. Comes with the default error handling * behavior. */ export const EMPTY_OBSERVER: Readonly<Observer<any>> & { closed: true } = { closed: true, next: noop, error: defaultErrorHandler, complete: noop, };