Skip to content

Commit

Permalink
refactor: Replace Zone.scheduleMacroTask with ExperimentalPendingTasks
Browse files Browse the repository at this point in the history
  • Loading branch information
mmalerba committed May 31, 2024
1 parent 19d3a34 commit 1b24707
Showing 1 changed file with 29 additions and 51 deletions.
80 changes: 29 additions & 51 deletions src/zones.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,10 @@
/* eslint-disable @typescript-eslint/ban-ts-comment */
import { Injectable, NgZone } from "@angular/core";
import {
ExperimentalPendingTasks,
Injectable,
NgZone,
inject,
} from "@angular/core";
import {
Observable,
Operator,
Expand Down Expand Up @@ -48,17 +53,16 @@ export class ɵZoneScheduler implements SchedulerLike {
}

class BlockUntilFirstOperator<T> implements Operator<T, T> {
// @ts-ignore
private task: MacroTask | null = null;

constructor(private zone: any) {}
constructor(
private zone: any,
private pendingTasks: ExperimentalPendingTasks
) {}

call(subscriber: Subscriber<T>, source: Observable<T>): TeardownLogic {
const unscheduleTask = this.unscheduleTask.bind(this);
// @ts-ignore
this.task = this.zone.run(() =>
Zone.current.scheduleMacroTask("firebaseZoneBlock", noop, {}, noop, noop)
);
const taskDone = this.zone.run(() => this.pendingTasks.add());
// maybe this is a race condition, invoke in a timeout
// hold for 10ms while I try to figure out what is going on
const unscheduleTask = () => setTimeout(taskDone, 10);

return source
.pipe(
Expand All @@ -71,17 +75,6 @@ class BlockUntilFirstOperator<T> implements Operator<T, T> {
.subscribe(subscriber)
.add(unscheduleTask);
}

private unscheduleTask() {
// maybe this is a race condition, invoke in a timeout
// hold for 10ms while I try to figure out what is going on
setTimeout(() => {
if (this.task != null && this.task.state === "scheduled") {
this.task.invoke();
this.task = null;
}
}, 10);
}
}

@Injectable({
Expand All @@ -90,14 +83,15 @@ class BlockUntilFirstOperator<T> implements Operator<T, T> {
export class ɵAngularFireSchedulers {
public readonly outsideAngular: ɵZoneScheduler;
public readonly insideAngular: ɵZoneScheduler;
public readonly pendingTasks = inject(ExperimentalPendingTasks);

constructor(public ngZone: NgZone) {
// @ts-ignore
this.outsideAngular = ngZone.runOutsideAngular(
// @ts-ignore
() => new ɵZoneScheduler(Zone.current)
);
// @ts-ignore
this.insideAngular = ngZone.run(
// @ts-ignore
() => new ɵZoneScheduler(Zone.current, asyncScheduler)
);
globalThis.ɵAngularFireScheduler ||= this;
Expand Down Expand Up @@ -149,7 +143,9 @@ export function ɵkeepUnstableUntilFirstFactory(
return function keepUnstableUntilFirst<T>(
obs$: Observable<T>
): Observable<T> {
obs$ = obs$.lift(new BlockUntilFirstOperator(schedulers.ngZone));
obs$ = obs$.lift(
new BlockUntilFirstOperator(schedulers.ngZone, schedulers.pendingTasks)
);

return obs$.pipe(
// Run the subscribe body outside of Angular (e.g. calling Firebase SDK to add a listener to a change event)
Expand All @@ -165,19 +161,15 @@ export function ɵkeepUnstableUntilFirstFactory(
// @ts-ignore
const zoneWrapFn = (
it: (...args: any[]) => any,
macrotask: MacroTask | undefined
taskDone: VoidFunction | undefined
) => {
// eslint-disable-next-line @typescript-eslint/no-this-alias
const _this = this;
// function() is needed for the arguments object
return function () {
const _arguments = arguments;
if (macrotask) {
setTimeout(() => {
if (macrotask.state === "scheduled") {
macrotask.invoke();
}
}, 10);
if (taskDone) {
setTimeout(taskDone, 10);
}
return run(() => it.apply(_this, _arguments));
};
Expand All @@ -186,27 +178,17 @@ const zoneWrapFn = (
export const ɵzoneWrap = <T = unknown>(it: T, blockUntilFirst: boolean): T => {
// function() is needed for the arguments object
return function () {
// @ts-ignore
let macrotask: MacroTask | undefined;
let taskDone: VoidFunction | undefined;
const _arguments = arguments;
// if this is a callback function, e.g, onSnapshot, we should create a microtask and invoke it
// if this is a callback function, e.g, onSnapshot, we should create a pending task and complete it
// only once one of the callback functions is tripped.
for (let i = 0; i < arguments.length; i++) {
if (typeof _arguments[i] === "function") {
if (blockUntilFirst) {
// @ts-ignore
macrotask ||= run(() =>
Zone.current.scheduleMacroTask(
"firebaseZoneBlock",
noop,
{},
noop,
noop
)
);
taskDone ||= run(() => getSchedulers().pendingTasks.add());
}
// TODO create a microtask to track callback functions
_arguments[i] = zoneWrapFn(_arguments[i], macrotask);
_arguments[i] = zoneWrapFn(_arguments[i], taskDone);
}
}
const ret = runOutsideAngular(() => (it as any).apply(this, _arguments));
Expand Down Expand Up @@ -234,15 +216,11 @@ export const ɵzoneWrap = <T = unknown>(it: T, blockUntilFirst: boolean): T => {
)
)
);
} else if (typeof ret === "function" && macrotask) {
} else if (typeof ret === "function" && taskDone) {
// Handle unsubscribe
// function() is needed for the arguments object
return function () {
setTimeout(() => {
if (macrotask && macrotask.state === "scheduled") {
macrotask.invoke();
}
}, 10);
setTimeout(taskDone, 10);
return ret.apply(this, arguments);
};
} else {
Expand Down

0 comments on commit 1b24707

Please sign in to comment.