Skip to content

Commit d017b3a

Browse files
committed
refactor(SafeSubscriber): optimize perf for ordinary observers
Related ReactiveX#6815
1 parent 7214daa commit d017b3a

File tree

1 file changed

+40
-41
lines changed

1 file changed

+40
-41
lines changed

src/internal/Subscriber.ts

Lines changed: 40 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -146,55 +146,54 @@ function bind<Fn extends (...args: any[]) => any>(fn: Fn, thisArg: any): Fn {
146146
return _bind.call(fn, thisArg);
147147
}
148148

149+
class ConsumerObserver<T> implements Observer<T> {
150+
constructor(private partialObserver: Partial<Observer<T>>) {}
151+
152+
next(value: T): void {
153+
if (this.partialObserver.next) {
154+
try {
155+
this.partialObserver.next(value);
156+
} catch (error) {
157+
reportUnhandledError(error);
158+
}
159+
}
160+
}
161+
162+
error(err: any): void {
163+
if (this.partialObserver.error) {
164+
try {
165+
this.partialObserver.error(err);
166+
} catch (error) {
167+
reportUnhandledError(error);
168+
}
169+
} else {
170+
reportUnhandledError(err);
171+
}
172+
}
173+
174+
complete(): void {
175+
if (this.partialObserver.complete) {
176+
try {
177+
this.partialObserver.complete();
178+
} catch (error) {
179+
reportUnhandledError(error);
180+
}
181+
}
182+
}
183+
}
184+
149185
export class SafeSubscriber<T> extends Subscriber<T> {
150186
constructor(observerOrNext?: Partial<Observer<T>> | ((value: T) => void) | null) {
151187
super();
152188

153-
let next: ((value: T) => void) | undefined;
154-
let error: ((err: any) => void) | undefined;
155-
let complete: (() => void) | undefined;
156-
if (isFunction(observerOrNext)) {
157-
// The first argument is a function, not an observer. The next
158-
// two arguments *could* be observers, or they could be empty.
159-
next = observerOrNext;
160-
} else if (observerOrNext) {
161-
// The first argument is an observer object, we have to pull the handlers
162-
// off and capture the owner object as the context. That is because we're
163-
// going to put them all in a new destination with ensured methods
164-
// for `next`, `error`, and `complete`. That's part of what makes this
165-
// the "Safe" Subscriber.
166-
({ next, error, complete } = observerOrNext);
167-
next = next && bind(next, observerOrNext);
168-
error = error && bind(error, observerOrNext);
169-
complete = complete && bind(complete, observerOrNext);
170-
}
189+
const partialObserver = !observerOrNext || isFunction(observerOrNext) ? { next: observerOrNext ?? undefined } : observerOrNext;
171190

172-
// Once we set the destination, the superclass `Subscriber` will
173-
// do it's magic in the `_next`, `_error`, and `_complete` methods.
174-
this.destination = {
175-
next: next ? wrapForErrorHandling(next, this) : noop,
176-
error: wrapForErrorHandling(error ?? defaultErrorHandler, this),
177-
complete: complete ? wrapForErrorHandling(complete, this) : noop,
178-
};
191+
// Wrap the partial observer to ensure it's a full observer, and
192+
// make sure proper error handling is accounted for.
193+
this.destination = new ConsumerObserver(partialObserver);
179194
}
180195
}
181196

182-
/**
183-
* Wraps a user-provided handler (or our {@link defaultErrorHandler} in one case) to
184-
* ensure that any thrown errors are caught and handled appropriately.
185-
*
186-
* @param handler The handler to wrap
187-
* @param instance The SafeSubscriber instance we're going to mark if there's an error.
188-
*/
189-
function wrapForErrorHandling(handler: (arg?: any) => void, instance: SafeSubscriber<any>) {
190-
return (...args: any[]) => {
191-
try {
192-
handler(...args);
193-
} catch (err) {
194-
reportUnhandledError(err);
195-
}
196-
};
197-
}
198197
/**
199198
* An error handler used when no error handler was supplied
200199
* to the SafeSubscriber -- meaning no error handler was supplied

0 commit comments

Comments
 (0)