rxjs - npm Package Compare versions

Comparing version 7.2.0 to 7.3.0

dist/cjs/internal/AsyncSubject.js

@@ -30,7 +30,7 @@ "use strict";
     AsyncSubject.prototype._checkFinalizedStatuses = function (subscriber) {
-        var _a = this, hasError = _a.hasError, _hasValue = _a._hasValue, _value = _a._value, thrownError = _a.thrownError, isStopped = _a.isStopped;
+        var _a = this, hasError = _a.hasError, _hasValue = _a._hasValue, _value = _a._value, thrownError = _a.thrownError, isStopped = _a.isStopped, _isComplete = _a._isComplete;
         if (hasError) {
             subscriber.error(thrownError);
         }
-        else if (isStopped) {
+        else if (isStopped || _isComplete) {
             _hasValue && subscriber.next(_value);
@@ -37,0 +37,0 @@ subscriber.complete();

dist/cjs/internal/Observable.js

@@ -75,3 +75,3 @@ "use strict";
         }
-        return operations.length ? pipe_1.pipeFromArray(operations)(this) : this;
+        return pipe_1.pipeFromArray(operations)(this);
     };
@@ -78,0 +78,0 @@ Observable.prototype.toPromise = function (promiseCtor) {

dist/cjs/internal/operators/retry.js

@@ -7,2 +7,4 @@ "use strict";
 var identity_1 = require("../util/identity");
+var timer_1 = require("../observable/timer");
+var from_1 = require("../observable/from");
 function retry(configOrCount) {
@@ -19,3 +21,3 @@ if (configOrCount === void 0) { configOrCount = Infinity; }
     }
-    var count = config.count, _a = config.resetOnSuccess, resetOnSuccess = _a === void 0 ? false : _a;
+    var _a = config.count, count = _a === void 0 ? Infinity : _a, delay = config.delay, _b = config.resetOnSuccess, resetOnSuccess = _b === void 0 ? false : _b;
     return count <= 0
@@ -35,9 +37,24 @@ ? identity_1.identity
                     if (soFar++ < count) {
-                        if (innerSub) {
-                            innerSub.unsubscribe();
-                            innerSub = null;
-                            subscribeForRetry();
+                        var resub_1 = function () {
+                            if (innerSub) {
+                                innerSub.unsubscribe();
+                                innerSub = null;
+                                subscribeForRetry();
+                            }
+                            else {
+                                syncUnsub = true;
+                            }
+                        };
+                        if (delay != null) {
+                            var notifier = typeof delay === 'number' ? timer_1.timer(delay) : from_1.innerFrom(delay(err, soFar));
+                            var notifierSubscriber_1 = new OperatorSubscriber_1.OperatorSubscriber(subscriber, function () {
+                                notifierSubscriber_1.unsubscribe();
+                                resub_1();
+                            }, function () {
+                                subscriber.complete();
+                            });
+                            notifier.subscribe(notifierSubscriber_1);
                         }
                         else {
-                            syncUnsub = true;
+                            resub_1();
                         }
@@ -44,0 +61,0 @@ }

dist/cjs/internal/operators/tap.js

@@ -9,5 +9,11 @@ "use strict";
 function tap(observerOrNext, error, complete) {
-    var tapObserver = isFunction_1.isFunction(observerOrNext) || error || complete ? { next: observerOrNext, error: error, complete: complete } : observerOrNext;
+    var tapObserver = isFunction_1.isFunction(observerOrNext) || error || complete
+        ?
+            { next: observerOrNext, error: error, complete: complete }
+        : observerOrNext;
     return tapObserver
         ? lift_1.operate(function (source, subscriber) {
+            var _a;
+            (_a = tapObserver.subscribe) === null || _a === void 0 ? void 0 : _a.call(tapObserver);
+            var isUnsub = true;
             source.subscribe(new OperatorSubscriber_1.OperatorSubscriber(subscriber, function (value) {
@@ -19,2 +25,3 @@ var _a;
                 var _a;
+                isUnsub = false;
                 (_a = tapObserver.complete) === null || _a === void 0 ? void 0 : _a.call(tapObserver);
@@ -24,4 +31,11 @@ subscriber.complete();
                 var _a;
+                isUnsub = false;
                 (_a = tapObserver.error) === null || _a === void 0 ? void 0 : _a.call(tapObserver, err);
                 subscriber.error(err);
+            }, function () {
+                var _a, _b;
+                if (isUnsub) {
+                    (_a = tapObserver.unsubscribe) === null || _a === void 0 ? void 0 : _a.call(tapObserver);
+                }
+                (_b = tapObserver.finalize) === null || _b === void 0 ? void 0 : _b.call(tapObserver);
             }));
@@ -28,0 +42,0 @@ })

dist/esm/internal/AsyncSubject.js

@@ -10,7 +10,7 @@ import { Subject } from './Subject';
     _checkFinalizedStatuses(subscriber) {
-        const { hasError, _hasValue, _value, thrownError, isStopped } = this;
+        const { hasError, _hasValue, _value, thrownError, isStopped, _isComplete } = this;
         if (hasError) {
             subscriber.error(thrownError);
         }
-        else if (isStopped) {
+        else if (isStopped || _isComplete) {
             _hasValue && subscriber.next(_value);
@@ -17,0 +17,0 @@ subscriber.complete();

dist/esm/internal/Observable.js

@@ -66,3 +66,3 @@ import { SafeSubscriber, Subscriber } from './Subscriber';
     pipe(...operations) {
-        return operations.length ? pipeFromArray(operations)(this) : this;
+        return pipeFromArray(operations)(this);
     }
@@ -69,0 +69,0 @@ toPromise(promiseCtor) {

dist/esm/internal/operators/retry.js

 import { operate } from '../util/lift';
 import { OperatorSubscriber } from './OperatorSubscriber';
 import { identity } from '../util/identity';
+import { timer } from '../observable/timer';
+import { innerFrom } from '../observable/from';
 export function retry(configOrCount = Infinity) {
@@ -14,3 +16,3 @@ let config;
     }
-    const { count, resetOnSuccess = false } = config;
+    const { count = Infinity, delay, resetOnSuccess: resetOnSuccess = false } = config;
     return count <= 0
@@ -30,9 +32,24 @@ ? identity
                     if (soFar++ < count) {
-                        if (innerSub) {
-                            innerSub.unsubscribe();
-                            innerSub = null;
-                            subscribeForRetry();
+                        const resub = () => {
+                            if (innerSub) {
+                                innerSub.unsubscribe();
+                                innerSub = null;
+                                subscribeForRetry();
+                            }
+                            else {
+                                syncUnsub = true;
+                            }
+                        };
+                        if (delay != null) {
+                            const notifier = typeof delay === 'number' ? timer(delay) : innerFrom(delay(err, soFar));
+                            const notifierSubscriber = new OperatorSubscriber(subscriber, () => {
+                                notifierSubscriber.unsubscribe();
+                                resub();
+                            }, () => {
+                                subscriber.complete();
+                            });
+                            notifier.subscribe(notifierSubscriber);
                         }
                         else {
-                            syncUnsub = true;
+                            resub();
                         }
@@ -39,0 +56,0 @@ }

dist/esm/internal/operators/tap.js

@@ -6,5 +6,11 @@ import { isFunction } from '../util/isFunction';
 export function tap(observerOrNext, error, complete) {
-    const tapObserver = isFunction(observerOrNext) || error || complete ? { next: observerOrNext, error, complete } : observerOrNext;
+    const tapObserver = isFunction(observerOrNext) || error || complete
+        ?
+            { next: observerOrNext, error, complete }
+        : observerOrNext;
     return tapObserver
         ? operate((source, subscriber) => {
+            var _a;
+            (_a = tapObserver.subscribe) === null || _a === void 0 ? void 0 : _a.call(tapObserver);
+            let isUnsub = true;
             source.subscribe(new OperatorSubscriber(subscriber, (value) => {
@@ -16,2 +22,3 @@ var _a;
                 var _a;
+                isUnsub = false;
                 (_a = tapObserver.complete) === null || _a === void 0 ? void 0 : _a.call(tapObserver);
@@ -21,4 +28,11 @@ subscriber.complete();
                 var _a;
+                isUnsub = false;
                 (_a = tapObserver.error) === null || _a === void 0 ? void 0 : _a.call(tapObserver, err);
                 subscriber.error(err);
+            }, () => {
+                var _a, _b;
+                if (isUnsub) {
+                    (_a = tapObserver.unsubscribe) === null || _a === void 0 ? void 0 : _a.call(tapObserver);
+                }
+                (_b = tapObserver.finalize) === null || _b === void 0 ? void 0 : _b.call(tapObserver);
             }));
@@ -25,0 +39,0 @@ })

dist/esm5/internal/AsyncSubject.js

@@ -13,7 +13,7 @@ import { __extends } from "tslib";
     AsyncSubject.prototype._checkFinalizedStatuses = function (subscriber) {
-        var _a = this, hasError = _a.hasError, _hasValue = _a._hasValue, _value = _a._value, thrownError = _a.thrownError, isStopped = _a.isStopped;
+        var _a = this, hasError = _a.hasError, _hasValue = _a._hasValue, _value = _a._value, thrownError = _a.thrownError, isStopped = _a.isStopped, _isComplete = _a._isComplete;
         if (hasError) {
             subscriber.error(thrownError);
         }
-        else if (isStopped) {
+        else if (isStopped || _isComplete) {
             _hasValue && subscriber.next(_value);
@@ -20,0 +20,0 @@ subscriber.complete();

dist/esm5/internal/Observable.js

@@ -72,3 +72,3 @@ import { SafeSubscriber, Subscriber } from './Subscriber';
         }
-        return operations.length ? pipeFromArray(operations)(this) : this;
+        return pipeFromArray(operations)(this);
     };
@@ -75,0 +75,0 @@ Observable.prototype.toPromise = function (promiseCtor) {

dist/esm5/internal/operators/retry.js

 import { operate } from '../util/lift';
 import { OperatorSubscriber } from './OperatorSubscriber';
 import { identity } from '../util/identity';
+import { timer } from '../observable/timer';
+import { innerFrom } from '../observable/from';
 export function retry(configOrCount) {
@@ -15,3 +17,3 @@ if (configOrCount === void 0) { configOrCount = Infinity; }
     }
-    var count = config.count, _a = config.resetOnSuccess, resetOnSuccess = _a === void 0 ? false : _a;
+    var _a = config.count, count = _a === void 0 ? Infinity : _a, delay = config.delay, _b = config.resetOnSuccess, resetOnSuccess = _b === void 0 ? false : _b;
     return count <= 0
@@ -31,9 +33,24 @@ ? identity
                     if (soFar++ < count) {
-                        if (innerSub) {
-                            innerSub.unsubscribe();
-                            innerSub = null;
-                            subscribeForRetry();
+                        var resub_1 = function () {
+                            if (innerSub) {
+                                innerSub.unsubscribe();
+                                innerSub = null;
+                                subscribeForRetry();
+                            }
+                            else {
+                                syncUnsub = true;
+                            }
+                        };
+                        if (delay != null) {
+                            var notifier = typeof delay === 'number' ? timer(delay) : innerFrom(delay(err, soFar));
+                            var notifierSubscriber_1 = new OperatorSubscriber(subscriber, function () {
+                                notifierSubscriber_1.unsubscribe();
+                                resub_1();
+                            }, function () {
+                                subscriber.complete();
+                            });
+                            notifier.subscribe(notifierSubscriber_1);
                         }
                         else {
-                            syncUnsub = true;
+                            resub_1();
                         }
@@ -40,0 +57,0 @@ }

dist/esm5/internal/operators/tap.js

@@ -6,5 +6,11 @@ import { isFunction } from '../util/isFunction';
 export function tap(observerOrNext, error, complete) {
-    var tapObserver = isFunction(observerOrNext) || error || complete ? { next: observerOrNext, error: error, complete: complete } : observerOrNext;
+    var tapObserver = isFunction(observerOrNext) || error || complete
+        ?
+            { next: observerOrNext, error: error, complete: complete }
+        : observerOrNext;
     return tapObserver
         ? operate(function (source, subscriber) {
+            var _a;
+            (_a = tapObserver.subscribe) === null || _a === void 0 ? void 0 : _a.call(tapObserver);
+            var isUnsub = true;
             source.subscribe(new OperatorSubscriber(subscriber, function (value) {
@@ -16,2 +22,3 @@ var _a;
                 var _a;
+                isUnsub = false;
                 (_a = tapObserver.complete) === null || _a === void 0 ? void 0 : _a.call(tapObserver);
@@ -21,4 +28,11 @@ subscriber.complete();
                 var _a;
+                isUnsub = false;
                 (_a = tapObserver.error) === null || _a === void 0 ? void 0 : _a.call(tapObserver, err);
                 subscriber.error(err);
+            }, function () {
+                var _a, _b;
+                if (isUnsub) {
+                    (_a = tapObserver.unsubscribe) === null || _a === void 0 ? void 0 : _a.call(tapObserver);
+                }
+                (_b = tapObserver.finalize) === null || _b === void 0 ? void 0 : _b.call(tapObserver);
             }));
@@ -25,0 +39,0 @@ })

dist/types/internal/observable/connectable.d.ts

@@ -1,17 +0,2 @@
-import { ObservableInput, SubjectLike } from '../types';
-import { Subscription } from '../Subscription';
-import { Observable } from '../Observable';
-/**
- * An observable with a `connect` method that is used to create a subscription
- * to an underlying source, connecting it with all consumers via a multicast.
- */
-export interface ConnectableObservableLike<T> extends Observable<T> {
-    /**
-     * (Idempotent) Calling this method will connect the underlying source observable to all subscribed consumers
-     * through an underlying {@link Subject}.
-     * @returns A subscription, that when unsubscribed, will "disconnect" the source from the connector subject,
-     * severing notifications to all consumers.
-     */
-    connect(): Subscription;
-}
+import { Connectable, ObservableInput, SubjectLike } from '../types';
 export interface ConnectableConfig<T> {
@@ -41,3 +26,3 @@ /**
  */
-export declare function connectable<T>(source: ObservableInput<T>, config?: ConnectableConfig<T>): ConnectableObservableLike<T>;
+export declare function connectable<T>(source: ObservableInput<T>, config?: ConnectableConfig<T>): Connectable<T>;
 //# sourceMappingURL=connectable.d.ts.map

dist/types/internal/operators/finalize.d.ts

@@ -49,2 +49,3 @@ import { MonoTypeOperatorFunction } from '../types';
  * // results:
+ * //   '[next] Called'
  * //   0
@@ -51,0 +52,0 @@ * //   '[finalize] Called'

dist/types/internal/operators/retry.d.ts

@@ -1,4 +0,19 @@
-import { MonoTypeOperatorFunction } from '../types';
+import { MonoTypeOperatorFunction, ObservableInput } from '../types';
 export interface RetryConfig {
-    count: number;
+    /**
+     * The maximum number of times to retry.
+     */
+    count?: number;
+    /**
+     * The number of milliseconds to delay before retrying, OR a function to
+     * return a notifier for delaying. If a function is returned, that function should
+     * return a notifier that, when it emits will retry the source. If the notifier
+     * completes _without_ emitting, the resulting observable will complete without error,
+     * if the notifier errors, the error will be pushed to the result.
+     */
+    delay?: number | ((error: any, retryCount: number) => ObservableInput<any>);
+    /**
+     * Whether or not to reset the retry counter when the retried subscription
+     * emits its first value.
+     */
     resetOnSuccess?: boolean;
@@ -47,4 +62,4 @@ }
  *
- * @param {number} count - Number of retry attempts before failing.
- * @param {boolean} resetOnSuccess - When set to `true` every successful emission will reset the error count
+ * @param count - Number of retry attempts before failing.
+ * @param resetOnSuccess - When set to `true` every successful emission will reset the error count
  * @return A function that returns an Observable that will resubscribe to the
@@ -54,3 +69,10 @@ * source stream when the source stream errors, at most `count` times.
 export declare function retry<T>(count?: number): MonoTypeOperatorFunction<T>;
+/**
+ * Returns an observable that mirrors the source observable unless it errors. If it errors, the source observable
+ * will be resubscribed to (or "retried") based on the configuration passed here. See documentation
+ * for {@link RetryConfig} for more details.
+ *
+ * @param config - The retry configuration
+ */
 export declare function retry<T>(config: RetryConfig): MonoTypeOperatorFunction<T>;
 //# sourceMappingURL=retry.d.ts.map

dist/types/internal/operators/tap.d.ts

 import { MonoTypeOperatorFunction, Observer } from '../types';
-export declare function tap<T>(observer?: Partial<Observer<T>>): MonoTypeOperatorFunction<T>;
+export interface TapObserver<T> extends Observer<T> {
+    subscribe: () => void;
+    unsubscribe: () => void;
+    finalize: () => void;
+}
+export declare function tap<T>(observer?: Partial<TapObserver<T>>): MonoTypeOperatorFunction<T>;
 export declare function tap<T>(next: (value: T) => void): MonoTypeOperatorFunction<T>;
@@ -4,0 +9,0 @@ /** @deprecated Instead of passing separate callback arguments, use an observer argument. Signatures taking separate callback arguments will be removed in v8. Details: https://rxjs.dev/deprecations/subscribe-arguments */

dist/types/internal/types.d.ts

@@ -251,3 +251,16 @@ /// <reference lib="esnext.asynciterable" />
 }
+/**
+ * An observable with a `connect` method that is used to create a subscription
+ * to an underlying source, connecting it with all consumers via a multicast.
+ */
+export interface Connectable<T> extends Observable<T> {
+    /**
+     * (Idempotent) Calling this method will connect the underlying source observable to all subscribed consumers
+     * through an underlying {@link Subject}.
+     * @returns A subscription, that when unsubscribed, will "disconnect" the source from the connector subject,
+     * severing notifications to all consumers.
+     */
+    connect(): Subscription;
+}
 export {};
 //# sourceMappingURL=types.d.ts.map

package.json

 {
   "name": "rxjs",
-  "version": "7.2.0",
+  "version": "7.3.0",
   "description": "Reactive Extensions for modern JavaScript",
@@ -5,0 +5,0 @@ "main": "./dist/cjs/index.js",

src/internal/AsyncSubject.ts

@@ -17,6 +17,6 @@ import { Subject } from './Subject';
   protected _checkFinalizedStatuses(subscriber: Subscriber<T>) {
-    const { hasError, _hasValue, _value, thrownError, isStopped } = this;
+    const { hasError, _hasValue, _value, thrownError, isStopped, _isComplete } = this;
     if (hasError) {
       subscriber.error(thrownError);
-    } else if (isStopped) {
+    } else if (isStopped || _isComplete) {
       _hasValue && subscriber.next(_value!);
@@ -23,0 +23,0 @@ subscriber.complete();

src/internal/Observable.ts

@@ -439,3 +439,3 @@ /**
   pipe(...operations: OperatorFunction<any, any>[]): Observable<any> {
-    return operations.length ? pipeFromArray(operations)(this) : this;
+    return pipeFromArray(operations)(this);
   }
@@ -442,0 +442,0 @@

src/internal/observable/connectable.ts

@@ -1,2 +0,2 @@
-import { ObservableInput, SubjectLike } from '../types';
+import { Connectable, ObservableInput, SubjectLike } from '../types';
 import { Subject } from '../Subject';
@@ -7,16 +7,2 @@ import { Subscription } from '../Subscription';
 
-/**
- * An observable with a `connect` method that is used to create a subscription
- * to an underlying source, connecting it with all consumers via a multicast.
- */
-export interface ConnectableObservableLike<T> extends Observable<T> {
-  /**
-   * (Idempotent) Calling this method will connect the underlying source observable to all subscribed consumers
-   * through an underlying {@link Subject}.
-   * @returns A subscription, that when unsubscribed, will "disconnect" the source from the connector subject,
-   * severing notifications to all consumers.
-   */
-  connect(): Subscription;
-}
-
 export interface ConnectableConfig<T> {
@@ -55,3 +41,3 @@ /**
  */
-export function connectable<T>(source: ObservableInput<T>, config: ConnectableConfig<T> = DEFAULT_CONFIG): ConnectableObservableLike<T> {
+export function connectable<T>(source: ObservableInput<T>, config: ConnectableConfig<T> = DEFAULT_CONFIG): Connectable<T> {
   // The subscription representing the connection.
@@ -58,0 +44,0 @@ let connection: Subscription | null = null;

src/internal/operators/finalize.ts

@@ -51,2 +51,3 @@ import { MonoTypeOperatorFunction } from '../types';
  * // results:
+ * //   '[next] Called'
  * //   0
@@ -53,0 +54,0 @@ * //   '[finalize] Called'

src/internal/operators/retry.ts

@@ -1,2 +0,2 @@
-import { MonoTypeOperatorFunction } from '../types';
+import { MonoTypeOperatorFunction, ObservableInput } from '../types';
 import { operate } from '../util/lift';
@@ -6,5 +6,22 @@ import { Subscription } from '../Subscription';
 import { identity } from '../util/identity';
+import { timer } from '../observable/timer';
+import { innerFrom } from '../observable/from';
 
 export interface RetryConfig {
-  count: number;
+  /**
+   * The maximum number of times to retry.
+   */
+  count?: number;
+  /**
+   * The number of milliseconds to delay before retrying, OR a function to
+   * return a notifier for delaying. If a function is returned, that function should
+   * return a notifier that, when it emits will retry the source. If the notifier
+   * completes _without_ emitting, the resulting observable will complete without error,
+   * if the notifier errors, the error will be pushed to the result.
+   */
+  delay?: number | ((error: any, retryCount: number) => ObservableInput<any>);
+  /**
+   * Whether or not to reset the retry counter when the retried subscription
+   * emits its first value.
+   */
   resetOnSuccess?: boolean;
@@ -54,4 +71,4 @@ }
  *
- * @param {number} count - Number of retry attempts before failing.
- * @param {boolean} resetOnSuccess - When set to `true` every successful emission will reset the error count
+ * @param count - Number of retry attempts before failing.
+ * @param resetOnSuccess - When set to `true` every successful emission will reset the error count
  * @return A function that returns an Observable that will resubscribe to the
@@ -61,3 +78,12 @@ * source stream when the source stream errors, at most `count` times.
 export function retry<T>(count?: number): MonoTypeOperatorFunction<T>;
+
+/**
+ * Returns an observable that mirrors the source observable unless it errors. If it errors, the source observable
+ * will be resubscribed to (or "retried") based on the configuration passed here. See documentation
+ * for {@link RetryConfig} for more details.
+ *
+ * @param config - The retry configuration
+ */
 export function retry<T>(config: RetryConfig): MonoTypeOperatorFunction<T>;
+
 export function retry<T>(configOrCount: number | RetryConfig = Infinity): MonoTypeOperatorFunction<T> {
@@ -72,3 +98,3 @@ let config: RetryConfig;
   }
-  const { count, resetOnSuccess = false } = config;
+  const { count = Infinity, delay, resetOnSuccess: resetOnSuccess = false } = config;
 
@@ -86,2 +112,3 @@ return count <= 0
               (value) => {
+                // If we're resetting on success
                 if (resetOnSuccess) {
@@ -96,10 +123,41 @@ soFar = 0;
                 if (soFar++ < count) {
-                  if (innerSub) {
-                    innerSub.unsubscribe();
-                    innerSub = null;
-                    subscribeForRetry();
+                  // We are still under our retry count
+                  const resub = () => {
+                    if (innerSub) {
+                      innerSub.unsubscribe();
+                      innerSub = null;
+                      subscribeForRetry();
+                    } else {
+                      syncUnsub = true;
+                    }
+                  };
+
+                  if (delay != null) {
+                    // The user specified a retry delay.
+                    // They gave us a number, use a timer, otherwise, it's a function,
+                    // and we're going to call it to get a notifier.
+                    const notifier = typeof delay === 'number' ? timer(delay) : innerFrom(delay(err, soFar));
+                    const notifierSubscriber = new OperatorSubscriber(
+                      subscriber,
+                      () => {
+                        // After we get the first notification, we
+                        // unsubscribe from the notifer, because we don't want anymore
+                        // and we resubscribe to the source.
+                        notifierSubscriber.unsubscribe();
+                        resub();
+                      },
+                      () => {
+                        // The notifier completed without emitting.
+                        // The author is telling us they want to complete.
+                        subscriber.complete();
+                      }
+                    );
+                    notifier.subscribe(notifierSubscriber);
                   } else {
-                    syncUnsub = true;
+                    // There was no notifier given. Just resub immediately.
+                    resub();
                   }
                 } else {
+                  // We're past our maximum number of retries.
+                  // Just send along the error.
                   subscriber.error(err);
@@ -106,0 +164,0 @@ }

src/internal/operators/share.ts

@@ -86,3 +86,3 @@ import { Observable } from '../Observable';
  * source.subscribe(x => console.log('subscription 1: ', x));
- * source.subscribe(x => console.log('subscription 1: ', x));
+ * source.subscribe(x => console.log('subscription 2: ', x));
  *
@@ -92,12 +92,12 @@ * // Logs:
  * // subscription 1:  0
- * // subscription 1:  0
+ * // subscription 2:  0
  * // Processing:  1
  * // subscription 1:  1
- * // subscription 1:  1
+ * // subscription 2:  1
  * // Processing:  2
  * // subscription 1:  4
- * // subscription 1:  4
+ * // subscription 2:  4
  * // Processing:  3
  * // subscription 1:  9
- * // subscription 1:  9
+ * // subscription 2:  9
  * // ... and so on
@@ -104,0 +104,0 @@ * ```

src/internal/operators/tap.ts

@@ -7,3 +7,9 @@ import { MonoTypeOperatorFunction, Observer } from '../types';
 
-export function tap<T>(observer?: Partial<Observer<T>>): MonoTypeOperatorFunction<T>;
+export interface TapObserver<T> extends Observer<T> {
+  subscribe: () => void;
+  unsubscribe: () => void;
+  finalize: () => void;
+}
+
+export function tap<T>(observer?: Partial<TapObserver<T>>): MonoTypeOperatorFunction<T>;
 export function tap<T>(next: (value: T) => void): MonoTypeOperatorFunction<T>;
@@ -110,3 +116,3 @@ /** @deprecated Instead of passing separate callback arguments, use an observer argument. Signatures taking separate callback arguments will be removed in v8. Details: https://rxjs.dev/deprecations/subscribe-arguments */
 export function tap<T>(
-  observerOrNext?: Partial<Observer<T>> | ((value: T) => void) | null,
+  observerOrNext?: Partial<TapObserver<T>> | ((value: T) => void) | null,
   error?: ((e: any) => void) | null,
@@ -119,7 +125,11 @@ complete?: (() => void) | null
   const tapObserver =
-    isFunction(observerOrNext) || error || complete ? { next: observerOrNext as (value: T) => void, error, complete } : observerOrNext;
+    isFunction(observerOrNext) || error || complete
+      ? // tslint:disable-next-line: no-object-literal-type-assertion
+        ({ next: observerOrNext as Exclude<typeof observerOrNext, Partial<TapObserver<T>>>, error, complete } as Partial<TapObserver<T>>)
+      : observerOrNext;
 
-  // TODO: Use `operate` function once this PR lands: https://github.com/ReactiveX/rxjs/pull/5742
   return tapObserver
     ? operate((source, subscriber) => {
+        tapObserver.subscribe?.();
+        let isUnsub = true;
         source.subscribe(
@@ -133,2 +143,3 @@ new OperatorSubscriber(
             () => {
+              isUnsub = false;
               tapObserver.complete?.();
@@ -138,4 +149,11 @@ subscriber.complete();
             (err) => {
+              isUnsub = false;
               tapObserver.error?.(err);
               subscriber.error(err);
+            },
+            () => {
+              if (isUnsub) {
+                tapObserver.unsubscribe?.();
+              }
+              tapObserver.finalize?.();
             }
@@ -142,0 +160,0 @@ )

src/internal/types.ts

@@ -311,1 +311,15 @@ // https://github.com/microsoft/TypeScript/issues/40462#issuecomment-689879308
 }
+
+/**
+ * An observable with a `connect` method that is used to create a subscription
+ * to an underlying source, connecting it with all consumers via a multicast.
+ */
+export interface Connectable<T> extends Observable<T> {
+  /**
+   * (Idempotent) Calling this method will connect the underlying source observable to all subscribed consumers
+   * through an underlying {@link Subject}.
+   * @returns A subscription, that when unsubscribed, will "disconnect" the source from the connector subject,
+   * severing notifications to all consumers.
+   */
+  connect(): Subscription;
+}

No alert changes

Improved metrics

Total package byte prevSize

4448277

increased by0.42%

Lines of code

56775

increased by0.4%

No dependency changes