Skip to content

Commit eb26cbc

Browse files
benleshbenlesh
authored
feat(tap): now supports subscribe, unsubscribe, and finalize handlers (#6527)
* feat(tap): Adds subscribe, unsubscribe, finalize handlers This adds a common request/task for RxJS users, which are three new handlers: - `subscribe`: fires on subscription to the source - `unsubscribe`: fires when the subscription to the result is unsubscribed from, but NOT if the source completes or errors - `finalize`: always fires on finalization, (basically equivalent to `finalize`) * chore: update api_guardian files * chore: Remove old TODO comment
1 parent 5f69795 commit eb26cbc

File tree

4 files changed

+109
-7
lines changed

4 files changed

+109
-7
lines changed

‎api_guard/dist/types/index.d.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -712,7 +712,7 @@ export declare function takeWhile<T, S extends T>(predicate: (value: T, index: n
712712
export declare function takeWhile<T, S extends T>(predicate: (value: T, index: number) => value is S, inclusive: false): OperatorFunction<T, S>;
713713
export declare function takeWhile<T>(predicate: (value: T, index: number) => boolean, inclusive?: boolean): MonoTypeOperatorFunction<T>;
714714

715-
export declare function tap<T>(observer?: Partial<Observer<T>>): MonoTypeOperatorFunction<T>;
715+
export declare function tap<T>(observer?: Partial<TapObserver<T>>): MonoTypeOperatorFunction<T>;
716716
export declare function tap<T>(next: (value: T) => void): MonoTypeOperatorFunction<T>;
717717
export declare function tap<T>(next?: ((value: T) => void) | null, error?: ((error: any) => void) | null, complete?: (() => void) | null): MonoTypeOperatorFunction<T>;
718718

‎api_guard/dist/types/operators/index.d.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -283,7 +283,7 @@ export declare function takeWhile<T, S extends T>(predicate: (value: T, index: n
283283
export declare function takeWhile<T, S extends T>(predicate: (value: T, index: number) => value is S, inclusive: false): OperatorFunction<T, S>;
284284
export declare function takeWhile<T>(predicate: (value: T, index: number) => boolean, inclusive?: boolean): MonoTypeOperatorFunction<T>;
285285

286-
export declare function tap<T>(observer?: Partial<Observer<T>>): MonoTypeOperatorFunction<T>;
286+
export declare function tap<T>(observer?: Partial<TapObserver<T>>): MonoTypeOperatorFunction<T>;
287287
export declare function tap<T>(next: (value: T) => void): MonoTypeOperatorFunction<T>;
288288
export declare function tap<T>(next?: ((value: T) => void) | null, error?: ((error: any) => void) | null, complete?: (() => void) | null): MonoTypeOperatorFunction<T>;
289289

‎spec/operators/tap-spec.ts

Lines changed: 85 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
/** @prettier */
22
import { expect } from 'chai';
33
import { tap, mergeMap, take } from 'rxjs/operators';
4-
import { Subject, of, throwError, Observer, EMPTY, Observable } from 'rxjs';
4+
import { Subject, of, throwError, Observer, EMPTY, Observable, noop } from 'rxjs';
55
import { TestScheduler } from 'rxjs/testing';
66
import { observableMatcher } from '../helpers/observableMatcher';
77

@@ -310,4 +310,88 @@ describe('tap', () => {
310310

311311
expect(sideEffects).to.deep.equal([0, 1, 2]);
312312
});
313+
314+
describe('lifecycle handlers', () => {
315+
it('should support an unsubscribe event that fires before finalize', () => {
316+
const results: any[] = [];
317+
const subject = new Subject<number>();
318+
319+
const subscription = subject
320+
.pipe(
321+
tap({
322+
subscribe: () => results.push('subscribe'),
323+
next: (value) => results.push(`next ${value}`),
324+
error: (err) => results.push(`error: ${err.message}`),
325+
complete: () => results.push('complete'),
326+
unsubscribe: () => results.push('unsubscribe'),
327+
finalize: () => results.push('finalize'),
328+
})
329+
)
330+
.subscribe();
331+
332+
subject.next(1);
333+
subject.next(2);
334+
expect(results).to.deep.equal(['subscribe', 'next 1', 'next 2']);
335+
336+
subscription.unsubscribe();
337+
338+
expect(results).to.deep.equal(['subscribe', 'next 1', 'next 2', 'unsubscribe', 'finalize']);
339+
});
340+
341+
it('should not call unsubscribe if source completes', () => {
342+
const results: any[] = [];
343+
const subject = new Subject<number>();
344+
345+
const subscription = subject
346+
.pipe(
347+
tap({
348+
subscribe: () => results.push('subscribe'),
349+
next: (value) => results.push(`next ${value}`),
350+
error: (err) => results.push(`error: ${err.message}`),
351+
complete: () => results.push('complete'),
352+
unsubscribe: () => results.push('unsubscribe'),
353+
finalize: () => results.push('finalize'),
354+
})
355+
)
356+
.subscribe();
357+
358+
subject.next(1);
359+
subject.next(2);
360+
expect(results).to.deep.equal(['subscribe', 'next 1', 'next 2']);
361+
subject.complete();
362+
// should have no effect
363+
subscription.unsubscribe();
364+
365+
expect(results).to.deep.equal(['subscribe', 'next 1', 'next 2', 'complete', 'finalize']);
366+
});
367+
368+
it('should not call unsubscribe if source errors', () => {
369+
const results: any[] = [];
370+
const subject = new Subject<number>();
371+
372+
const subscription = subject
373+
.pipe(
374+
tap({
375+
subscribe: () => results.push('subscribe'),
376+
next: (value) => results.push(`next ${value}`),
377+
error: (err) => results.push(`error: ${err.message}`),
378+
complete: () => results.push('complete'),
379+
unsubscribe: () => results.push('unsubscribe'),
380+
finalize: () => results.push('finalize'),
381+
})
382+
)
383+
.subscribe({
384+
error: noop,
385+
});
386+
387+
subject.next(1);
388+
subject.next(2);
389+
expect(results).to.deep.equal(['subscribe', 'next 1', 'next 2']);
390+
subject.error(new Error('bad'));
391+
// should have no effect
392+
subscription.unsubscribe();
393+
394+
expect(results).to.deep.equal(['subscribe', 'next 1', 'next 2', 'error: bad', 'finalize']);
395+
});
396+
});
313397
});

‎src/internal/operators/tap.ts

Lines changed: 22 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,13 @@ import { operate } from '../util/lift';
44
import { OperatorSubscriber } from './OperatorSubscriber';
55
import { identity } from '../util/identity';
66

7-
export function tap<T>(observer?: Partial<Observer<T>>): MonoTypeOperatorFunction<T>;
7+
export interface TapObserver<T> extends Observer<T> {
8+
subscribe: () => void;
9+
unsubscribe: () => void;
10+
finalize: () => void;
11+
}
12+
13+
export function tap<T>(observer?: Partial<TapObserver<T>>): MonoTypeOperatorFunction<T>;
814
export function tap<T>(next: (value: T) => void): MonoTypeOperatorFunction<T>;
915
/** @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 */
1016
export function tap<T>(
@@ -106,19 +112,23 @@ export function tap<T>(
106112
* runs the specified Observer or callback(s) for each item.
107113
*/
108114
export function tap<T>(
109-
observerOrNext?: Partial<Observer<T>> | ((value: T) => void) | null,
115+
observerOrNext?: Partial<TapObserver<T>> | ((value: T) => void) | null,
110116
error?: ((e: any) => void) | null,
111117
complete?: (() => void) | null
112118
): MonoTypeOperatorFunction<T> {
113119
// We have to check to see not only if next is a function,
114120
// but if error or complete were passed. This is because someone
115121
// could technically call tap like `tap(null, fn)` or `tap(null, null, fn)`.
116122
const tapObserver =
117-
isFunction(observerOrNext) || error || complete ? { next: observerOrNext as (value: T) => void, error, complete } : observerOrNext;
123+
isFunction(observerOrNext) || error || complete
124+
? // tslint:disable-next-line: no-object-literal-type-assertion
125+
({ next: observerOrNext as Exclude<typeof observerOrNext, Partial<TapObserver<T>>>, error, complete } as Partial<TapObserver<T>>)
126+
: observerOrNext;
118127

119-
// TODO: Use `operate` function once this PR lands: https://github.com/ReactiveX/rxjs/pull/5742
120128
return tapObserver
121129
? operate((source, subscriber) => {
130+
tapObserver.subscribe?.();
131+
let isUnsub = true;
122132
source.subscribe(
123133
new OperatorSubscriber(
124134
subscriber,
@@ -127,12 +137,20 @@ export function tap<T>(
127137
subscriber.next(value);
128138
},
129139
() => {
140+
isUnsub = false;
130141
tapObserver.complete?.();
131142
subscriber.complete();
132143
},
133144
(err) => {
145+
isUnsub = false;
134146
tapObserver.error?.(err);
135147
subscriber.error(err);
148+
},
149+
() => {
150+
if (isUnsub) {
151+
tapObserver.unsubscribe?.();
152+
}
153+
tapObserver.finalize?.();
136154
}
137155
)
138156
);

0 commit comments

Comments
 (0)