Skip to content

[Auth] Build state listeners on streams #7430

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 1 commit into from
Closed

[Auth] Build state listeners on streams #7430

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions common/api-review/util.api.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,11 @@ export function contains<T extends object>(obj: T, key: string): boolean;
// @public (undocumented)
export function createMockUserToken(token: EmulatorMockTokenOptions, projectId?: string): string;

// Warning: (ae-missing-release-tag) "createObserver" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal)
//
// @public (undocumented)
export function createObserver<T>(nextOrObserver?: NextFn<T> | PartialObserver<T>, error?: ErrorFn, complete?: CompleteFn): Observer<T>;

// Warning: (ae-missing-release-tag) "createSubscribe" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal)
//
// @public
Expand Down
167 changes: 120 additions & 47 deletions packages/auth/src/core/auth/auth_impl.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,10 @@ import {
FirebaseError,
getModularInstance,
Observer,
Subscribe
Subscribe,
Observable,
PartialObserver,
createObserver
} from '@firebase/util';

import { AuthInternal, ConfigInternal } from '../../model/auth';
Expand Down Expand Up @@ -84,8 +87,8 @@ export class AuthImpl implements AuthInternal, _FirebaseService {
private operations = Promise.resolve();
private persistenceManager?: PersistenceUserManager;
private redirectPersistenceManager?: PersistenceUserManager;
private authStateSubscription = new Subscription<User>(this);
private idTokenSubscription = new Subscription<User>(this);
private authStateEmitter = new EmitterStream<User | null>();
private idTokenEmitter = new EmitterStream<User | null>();
private readonly beforeStateQueue = new AuthMiddlewareQueue(this);
private redirectUser: UserInternal | null = null;
private isProactiveRefreshEnabled = false;
Expand Down Expand Up @@ -439,12 +442,9 @@ export class AuthImpl implements AuthInternal, _FirebaseService {
error?: ErrorFn,
completed?: CompleteFn
): Unsubscribe {
return this.registerStateListener(
this.authStateSubscription,
nextOrObserver,
error,
completed
);
const stream = this.createUserStateStream(this.authStateEmitter);
const observer: Observer<User | null> = createObserver(nextOrObserver, error, completed);
return stream.subscribe(observer);
}

beforeAuthStateChanged(
Expand All @@ -459,12 +459,9 @@ export class AuthImpl implements AuthInternal, _FirebaseService {
error?: ErrorFn,
completed?: CompleteFn
): Unsubscribe {
return this.registerStateListener(
this.idTokenSubscription,
nextOrObserver,
error,
completed
);
const stream = this.createUserStateStream(this.idTokenEmitter);
const observer: Observer<User | null> = createObserver(nextOrObserver, error, completed);
return stream.subscribe(observer);
}

toJSON(): object {
Expand Down Expand Up @@ -567,43 +564,35 @@ export class AuthImpl implements AuthInternal, _FirebaseService {
return;
}

this.idTokenSubscription.next(this.currentUser);
this.idTokenEmitter.emit(this.currentUser);

const currentUid = this.currentUser?.uid ?? null;
if (this.lastNotifiedUid !== currentUid) {
this.lastNotifiedUid = currentUid;
this.authStateSubscription.next(this.currentUser);
this.authStateEmitter.emit(this.currentUser);
}
}

private registerStateListener(
subscription: Subscription<User>,
nextOrObserver: NextOrObserver<User>,
error?: ErrorFn,
completed?: CompleteFn
): Unsubscribe {
private createUserStateStream(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for creating this PR and for the thorough analysis!

I wonder if a smaller scoped change like below would work (i have not tried this, just thinking out aloud). I also posted this on the issue:

private registerStateListener(
subscription: Subscription,
nextOrObserver: NextOrObserver,
error?: ErrorFn,
completed?: CompleteFn
): Unsubscribe {
if (this._deleted) {
return () => {};
}

  • let isUnsubscribed = false;
    const cb =
    typeof nextOrObserver === 'function'
    ? nextOrObserver
    : nextOrObserver.next.bind(nextOrObserver);

    const promise = this._isInitialized
    ? Promise.resolve()
    : this._initializationPromise;
    _assert(promise, this, AuthErrorCode.INTERNAL_ERROR);
    // The callback needs to be called asynchronously per the spec.
    // eslint-disable-next-line @typescript-eslint/no-floating-promises

    • promise.then(() => {if (isUnsubscribed) { return; }; cb(this.currentUser)});

    if (typeof nextOrObserver === 'function') {
    + Unsubscribe unsubscribe = subscription.addObserver(nextOrObserver, error, completed);
    + return {isUnsubscribed = true; unsubscribe();}
    } else {
    + Unsubscribe unsubscribe = subscription.addObserver(nextOrObserver);
    + return {isUnsubscribed = true; unsubscribe();}
    }
    }


Since the Observer logic is also used outside of Auth (in storage and firestore), i wonder if we can do a smaller fix. Thanks!

emitter: EmitterStream<User | null>
): Observable<User | null> {
if (this._deleted) {
return () => {};
return new NeverStream();
}

const cb =
typeof nextOrObserver === 'function'
? nextOrObserver
: nextOrObserver.next.bind(nextOrObserver);

const promise = this._isInitialized
? Promise.resolve()
: this._initializationPromise;
_assert(promise, this, AuthErrorCode.INTERNAL_ERROR);
// The callback needs to be called asynchronously per the spec.
// eslint-disable-next-line @typescript-eslint/no-floating-promises
promise.then(() => cb(this.currentUser));

if (typeof nextOrObserver === 'function') {
return subscription.addObserver(nextOrObserver, error, completed);
} else {
return subscription.addObserver(nextOrObserver);
}
const user = new PromiseStream(
promise.then(() => this.currentUser)
);

return new ConcatStream(
user,
emitter
);
}

/**
Expand Down Expand Up @@ -717,17 +706,101 @@ export function _castAuth(auth: Auth): AuthInternal {
return getModularInstance(auth) as AuthInternal;
}

/** Helper class to wrap subscriber logic */
class Subscription<T> {
private observer: Observer<T | null> | null = null;
readonly addObserver: Subscribe<T | null> = createSubscribe(
observer => (this.observer = observer)
);
class NeverStream<T> implements Observable<T> {
constructor() {}

subscribe(
nextOrObserver?: NextFn<T> | PartialObserver<T>,
error?: ErrorFn,
complete?: CompleteFn
): Unsubscribe {
return () => {};
}
}

class EmitterStream<T> implements Observable<T> {
constructor() {
this.subscribe = createSubscribe(
(o) => { this.observer = o; }
);
}

constructor(readonly auth: AuthInternal) {}
subscribe: Subscribe<T>;
observer?: Observer<T>;

get next(): NextFn<T | null> {
_assert(this.observer, this.auth, AuthErrorCode.INTERNAL_ERROR);
return this.observer.next.bind(this.observer);
emit(value: T) {
this.observer!.next(value);
}
}

class PromiseStream<T> implements Observable<T> {
constructor(promise: Promise<T>) {
this.promise = promise;
}

promise: Promise<T>;

subscribe(
nextOrObserver?: NextFn<T> | PartialObserver<T>,
error?: ErrorFn,
complete?: CompleteFn
): Unsubscribe {
const observer = createObserver(nextOrObserver, error, complete);

let isUnsubscribed = false;

this.promise.then(
(value) => {
if (isUnsubscribed) { return; }
observer.next(value);
observer.complete();
},
(error) => {
if (isUnsubscribed) { return; }
observer.error(error);
}
)

return () => {
isUnsubscribed = true;
}
}
}

class ConcatStream<T> implements Observable<T> {
constructor(s0: Observable<T>, s1: Observable<T>) {
this.s0 = s0;
this.s1 = s1;
}

s0: Observable<T>;
s1: Observable<T>;

subscribe(
nextOrObserver?: NextFn<T> | PartialObserver<T>,
error?: ErrorFn,
complete?: CompleteFn
): Unsubscribe {
const observer = createObserver(nextOrObserver, error, complete);

let unsubscribe0: Unsubscribe;
let unsubscribe1: Unsubscribe | null;

unsubscribe0 = this.s0.subscribe(
(value: T) => { observer.next(value); },
(error: Error) => { observer.error(error); },
() => {
unsubscribe1 = this.s1.subscribe(
(value: T) => { observer.next(value); },
(error: Error) => { observer.error(error); },
() => { observer.complete(); }
);
}
)

return () => {
unsubscribe1?.call(undefined);
unsubscribe0();
}
}
}
84 changes: 48 additions & 36 deletions packages/util/src/subscribe.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,51 @@ export function createSubscribe<T>(
return proxy.subscribe.bind(proxy);
}

export function createObserver<T>(
nextOrObserver?: NextFn<T> | PartialObserver<T>,
error?: ErrorFn,
complete?: CompleteFn
): Observer<T> {
let observer: Observer<T>;

if (
nextOrObserver === undefined &&
error === undefined &&
complete === undefined
) {
throw new Error('Missing Observer.');
}

// Assemble an Observer object when passed as callback functions.
if (
implementsAnyMethods(nextOrObserver as { [key: string]: unknown }, [
'next',
'error',
'complete'
])
) {
observer = nextOrObserver as Observer<T>;
} else {
observer = {
next: nextOrObserver as NextFn<T>,
error,
complete
} as Observer<T>;
}

if (observer.next === undefined) {
observer.next = noop as NextFn<T>;
}
if (observer.error === undefined) {
observer.error = noop as ErrorFn;
}
if (observer.complete === undefined) {
observer.complete = noop as CompleteFn;
}

return observer;
}

/**
* Implement fan-out for any number of Observers attached via a subscribe
* function.
Expand Down Expand Up @@ -130,42 +175,9 @@ class ObserverProxy<T> implements Observer<T> {
error?: ErrorFn,
complete?: CompleteFn
): Unsubscribe {
let observer: Observer<T>;

if (
nextOrObserver === undefined &&
error === undefined &&
complete === undefined
) {
throw new Error('Missing Observer.');
}

// Assemble an Observer object when passed as callback functions.
if (
implementsAnyMethods(nextOrObserver as { [key: string]: unknown }, [
'next',
'error',
'complete'
])
) {
observer = nextOrObserver as Observer<T>;
} else {
observer = {
next: nextOrObserver as NextFn<T>,
error,
complete
} as Observer<T>;
}

if (observer.next === undefined) {
observer.next = noop as NextFn<T>;
}
if (observer.error === undefined) {
observer.error = noop as ErrorFn;
}
if (observer.complete === undefined) {
observer.complete = noop as CompleteFn;
}
const observer = createObserver(
nextOrObserver, error, complete
);

const unsub = this.unsubscribeOne.bind(this, this.observers!.length);

Expand Down