diff --git a/packages/core/src/api/apis/implementations/AppThemeSelector/AppThemeSelector.ts b/packages/core/src/api/apis/implementations/AppThemeSelector/AppThemeSelector.ts index cc1146b336..7f6659e641 100644 --- a/packages/core/src/api/apis/implementations/AppThemeSelector/AppThemeSelector.ts +++ b/packages/core/src/api/apis/implementations/AppThemeSelector/AppThemeSelector.ts @@ -14,8 +14,9 @@ * limitations under the License. */ -import Observable from 'zen-observable'; import { AppThemeApi, AppTheme } from '../../definitions'; +import { BehaviorSubject } from '../lib'; +import { Observable } from '../../../types'; const STORAGE_KEY = 'theme'; @@ -50,34 +51,17 @@ export class AppThemeSelector implements AppThemeApi { return selector; } - private readonly themes: AppTheme[]; - private activeThemeId: string | undefined; + private readonly subject = new BehaviorSubject(undefined); - private readonly observable: Observable; - private readonly subscribers = new Set< - ZenObservable.SubscriptionObserver - >(); - - constructor(themes: AppTheme[]) { - this.themes = themes; - - this.observable = new Observable((subscriber) => { - subscriber.next(this.activeThemeId); - - this.subscribers.add(subscriber); - return () => { - this.subscribers.delete(subscriber); - }; - }); - } + constructor(private readonly themes: AppTheme[]) {} getInstalledThemes(): AppTheme[] { return this.themes.slice(); } activeThemeId$(): Observable { - return this.observable; + return this.subject; } getActiveThemeId(): string | undefined { @@ -86,6 +70,6 @@ export class AppThemeSelector implements AppThemeApi { setActiveThemeId(themeId?: string): void { this.activeThemeId = themeId; - this.subscribers.forEach((subscriber) => subscriber.next(themeId)); + this.subject.next(themeId); } } diff --git a/packages/core/src/api/apis/implementations/OAuthRequestManager/OAuthPendingRequests.ts b/packages/core/src/api/apis/implementations/OAuthRequestManager/OAuthPendingRequests.ts index 6348262022..96aada1864 100644 --- a/packages/core/src/api/apis/implementations/OAuthRequestManager/OAuthPendingRequests.ts +++ b/packages/core/src/api/apis/implementations/OAuthRequestManager/OAuthPendingRequests.ts @@ -14,8 +14,9 @@ * limitations under the License. */ -import Observable from 'zen-observable'; import { OAuthScopes } from '../../definitions'; +import { BehaviorSubject } from '../lib'; +import { Observable } from '../../../types'; type RequestQueueEntry = { scopes: OAuthScopes; @@ -44,16 +45,15 @@ export type OAuthPendingRequestsApi = { export class OAuthPendingRequests implements OAuthPendingRequestsApi { private requests: RequestQueueEntry[] = []; - private listeners: ZenObservable.SubscriptionObserver< - PendingRequest - >[] = []; + private subject = new BehaviorSubject>( + this.getCurrentPending(), + ); request(scopes: OAuthScopes): Promise { return new Promise((resolve, reject) => { this.requests.push({ scopes, resolve, reject }); - const pending = this.getCurrentPending(); - this.listeners.forEach((listener) => listener.next(pending)); + this.subject.next(this.getCurrentPending()); }); } @@ -66,26 +66,18 @@ export class OAuthPendingRequests return true; }); - const pending = this.getCurrentPending(); - this.listeners.forEach((listener) => listener.next(pending)); + this.subject.next(this.getCurrentPending()); } reject(error: Error) { this.requests.forEach((request) => request.reject(error)); this.requests = []; - const pending = this.getCurrentPending(); - this.listeners.forEach((listener) => listener.next(pending)); + this.subject.next(this.getCurrentPending()); } pending(): Observable> { - return new Observable((subscriber) => { - this.listeners.push(subscriber); - subscriber.next(this.getCurrentPending()); - return () => { - this.listeners = this.listeners.filter((l) => l !== subscriber); - }; - }); + return this.subject; } private getCurrentPending(): PendingRequest { diff --git a/packages/core/src/api/apis/implementations/OAuthRequestManager/OAuthRequestManager.ts b/packages/core/src/api/apis/implementations/OAuthRequestManager/OAuthRequestManager.ts index 02d89b9546..632ec059eb 100644 --- a/packages/core/src/api/apis/implementations/OAuthRequestManager/OAuthRequestManager.ts +++ b/packages/core/src/api/apis/implementations/OAuthRequestManager/OAuthRequestManager.ts @@ -21,8 +21,9 @@ import { AuthRequester, AuthRequesterOptions, } from '../../definitions'; -import Observable from 'zen-observable'; import { OAuthPendingRequests, PendingRequest } from './OAuthPendingRequests'; +import { BehaviorSubject } from '../lib'; +import { Observable } from '../../../types'; /** * The OAuthRequestManager is an implementation of the OAuthRequestApi. @@ -32,24 +33,10 @@ import { OAuthPendingRequests, PendingRequest } from './OAuthPendingRequests'; * them all together into a single request for each OAuth provider. */ export class OAuthRequestManager implements OAuthRequestApi { - private readonly request$: Observable; - private readonly observers = new Set< - ZenObservable.SubscriptionObserver - >(); + private readonly subject = new BehaviorSubject([]); private currentRequests: PendingAuthRequest[] = []; private handlerCount = 0; - constructor() { - this.request$ = new Observable((observer) => { - observer.next(this.currentRequests); - - this.observers.add(observer); - return () => { - this.observers.delete(observer); - }; - }).map((requests) => requests.filter(Boolean)); // Convert from sparse array to array of present items only - } - createAuthRequester(options: AuthRequesterOptions): AuthRequester { const handler = new OAuthPendingRequests(); @@ -66,7 +53,8 @@ export class OAuthRequestManager implements OAuthRequestApi { newRequests[index] = request; } this.currentRequests = newRequests; - this.observers.forEach((observer) => observer.next(newRequests)); + // Convert from sparse array to array of present items only + this.subject.next(newRequests.filter(Boolean)); }, }); @@ -100,7 +88,7 @@ export class OAuthRequestManager implements OAuthRequestApi { } authRequest$(): Observable { - return this.request$; + return this.subject; } async showLoginPopup(options: LoginPopupOptions): Promise { diff --git a/packages/core/src/api/apis/implementations/lib/index.ts b/packages/core/src/api/apis/implementations/lib/index.ts new file mode 100644 index 0000000000..aa2202858c --- /dev/null +++ b/packages/core/src/api/apis/implementations/lib/index.ts @@ -0,0 +1,17 @@ +/* + * Copyright 2020 Spotify AB + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +export * from './subjects'; diff --git a/packages/core/src/api/apis/implementations/lib/subjects.test.ts b/packages/core/src/api/apis/implementations/lib/subjects.test.ts new file mode 100644 index 0000000000..31ed26d97d --- /dev/null +++ b/packages/core/src/api/apis/implementations/lib/subjects.test.ts @@ -0,0 +1,178 @@ +/* + * Copyright 2020 Spotify AB + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { PublishSubject, BehaviorSubject } from './subjects'; + +function observerSpy() { + return { + next: jest.fn(), + error: jest.fn(), + complete: jest.fn(), + }; +} + +describe('PublishSubject', () => { + it('should be completed', async () => { + const subj = new PublishSubject(); + subj.complete(); + + const spy = observerSpy(); + subj.subscribe(spy); + await 'a tick'; + + expect(spy.next).not.toHaveBeenCalled(); + expect(spy.error).not.toHaveBeenCalled(); + expect(spy.complete).toHaveBeenCalledTimes(1); + + expect(() => subj.next(1)).toThrow('PublishSubject is closed'); + expect(() => subj.error(new Error())).toThrow('PublishSubject is closed'); + expect(() => subj.complete()).toThrow('PublishSubject is closed'); + + expect(subj.closed).toBe(true); + }); + + it('should forward values to all subscribers', async () => { + const subj = new PublishSubject(); + + const spy1 = observerSpy(); + const spy2 = observerSpy(); + subj.subscribe(spy1); + const sub = subj.subscribe(spy2); + + subj.next(42); + + expect(spy1.next).toHaveBeenCalledTimes(1); + expect(spy1.next).toHaveBeenCalledWith(42); + expect(spy2.next).toHaveBeenCalledTimes(1); + expect(spy2.next).toHaveBeenCalledWith(42); + + sub.unsubscribe(); + + subj.next(1337); + + expect(spy1.next).toHaveBeenCalledTimes(2); + expect(spy1.next).toHaveBeenCalledWith(1337); + expect(spy2.next).toHaveBeenCalledTimes(1); + + expect(spy1.error).not.toHaveBeenCalled(); + expect(spy2.error).not.toHaveBeenCalled(); + expect(spy1.complete).not.toHaveBeenCalled(); + expect(spy2.complete).not.toHaveBeenCalled(); + + expect(subj.closed).toBe(false); + }); + + it('should forward errors', async () => { + const subj = new PublishSubject(); + + const spy1 = observerSpy(); + subj.subscribe(spy1); + + const error = new Error('NOPE'); + subj.error(error); + expect(spy1.error).toHaveBeenCalledWith(error); + expect(spy1.next).not.toHaveBeenCalled(); + expect(spy1.complete).not.toHaveBeenCalled(); + + const spy2 = observerSpy(); + subj.subscribe(spy2); + await 'a tick'; + expect(spy2.error).toHaveBeenCalledWith(error); + expect(spy2.next).not.toHaveBeenCalled(); + expect(spy2.complete).not.toHaveBeenCalled(); + + expect(subj.closed).toBe(true); + }); +}); + +describe('BehaviorSubject', () => { + it('should be completed', async () => { + const subj = new BehaviorSubject(0); + subj.complete(); + + const next = jest.fn(); + const complete = jest.fn(); + subj.subscribe({ next, complete }); + await 'a tick'; + + expect(complete).toHaveBeenCalledTimes(1); + + expect(() => subj.next(1)).toThrow('BehaviorSubject is closed'); + expect(() => subj.error(new Error())).toThrow('BehaviorSubject is closed'); + expect(() => subj.complete()).toThrow('BehaviorSubject is closed'); + + expect(subj.closed).toBe(true); + }); + + it('should forward values to all subscribers', async () => { + const subj = new BehaviorSubject(0); + + const obs1 = jest.fn(); + const obs2 = jest.fn(); + subj.subscribe(obs1); + const sub = subj.subscribe(obs2); + await 'a tick'; + + expect(obs1).toHaveBeenCalledTimes(1); + expect(obs1).toHaveBeenCalledWith(0); + expect(obs2).toHaveBeenCalledTimes(1); + expect(obs2).toHaveBeenCalledWith(0); + + subj.next(42); + + expect(obs1).toHaveBeenCalledTimes(2); + expect(obs1).toHaveBeenCalledWith(42); + expect(obs2).toHaveBeenCalledTimes(2); + expect(obs2).toHaveBeenCalledWith(42); + + sub.unsubscribe(); + + subj.next(1337); + + expect(obs1).toHaveBeenCalledTimes(3); + expect(obs1).toHaveBeenCalledWith(1337); + expect(obs2).toHaveBeenCalledTimes(2); + + expect(subj.closed).toBe(false); + }); + + it('should forward errors', async () => { + const subj = new BehaviorSubject(0); + + const spy1 = observerSpy(); + subj.subscribe(spy1); + await 'a tick'; + + expect(spy1.error).not.toHaveBeenCalled(); + expect(spy1.next).toHaveBeenCalledWith(0); + + const error = new Error('NOPE'); + subj.error(error); + expect(spy1.error).toHaveBeenCalledWith(error); + expect(spy1.next).toHaveBeenCalledWith(0); + expect(spy1.next).toHaveBeenCalledTimes(1); + expect(spy1.complete).not.toHaveBeenCalled(); + + const spy2 = observerSpy(); + subj.subscribe(spy2); + await 'a tick'; + expect(spy2.error).toHaveBeenCalledWith(error); + expect(spy2.next).not.toHaveBeenCalled(); + expect(spy2.complete).not.toHaveBeenCalled(); + + expect(subj.closed).toBe(true); + }); +}); diff --git a/packages/core/src/api/apis/implementations/lib/subjects.ts b/packages/core/src/api/apis/implementations/lib/subjects.ts new file mode 100644 index 0000000000..b33df70a65 --- /dev/null +++ b/packages/core/src/api/apis/implementations/lib/subjects.ts @@ -0,0 +1,202 @@ +/* + * Copyright 2020 Spotify AB + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { Observable } from '../../../types'; +import ObservableImpl from 'zen-observable'; + +// TODO(Rugvip): These are stopgap and probably incomplete implementations of subjects. +// If we add a more complete Observables library they should be replaced. + +/** + * A basic implementation of ReactiveX publish subjects. + * + * A subject is a convenient way to create an observable when you want + * to fan out a single value to all subscribers. + * + * See http://reactivex.io/documentation/subject.html + */ +export class PublishSubject + implements Observable, ZenObservable.SubscriptionObserver { + private isClosed = false; + private terminatingError?: Error; + + private readonly observable = new ObservableImpl((subscriber) => { + if (this.isClosed) { + if (this.terminatingError) { + subscriber.error(this.terminatingError); + } else { + subscriber.complete(); + } + return () => {}; + } + + this.subscribers.add(subscriber); + return () => { + this.subscribers.delete(subscriber); + }; + }); + + private readonly subscribers = new Set< + ZenObservable.SubscriptionObserver + >(); + + get closed() { + return this.isClosed; + } + + next(value: T) { + if (this.isClosed) { + throw new Error('PublishSubject is closed'); + } + this.subscribers.forEach((subscriber) => subscriber.next(value)); + } + + error(error: Error) { + if (this.isClosed) { + throw new Error('PublishSubject is closed'); + } + this.isClosed = true; + this.terminatingError = error; + this.subscribers.forEach((subscriber) => subscriber.error(error)); + } + + complete() { + if (this.isClosed) { + throw new Error('PublishSubject is closed'); + } + this.isClosed = true; + this.subscribers.forEach((subscriber) => subscriber.complete()); + } + + subscribe(observer: ZenObservable.Observer): ZenObservable.Subscription; + subscribe( + onNext: (value: T) => void, + onError?: (error: any) => void, + onComplete?: () => void, + ): ZenObservable.Subscription; + subscribe( + onNext: ZenObservable.Observer | ((value: T) => void), + onError?: (error: any) => void, + onComplete?: () => void, + ): ZenObservable.Subscription { + const observer = + typeof onNext === 'function' + ? { + next: onNext, + error: onError, + complete: onComplete, + } + : onNext; + + return this.observable.subscribe(observer); + } +} + +/** + * A basic implementation of ReactiveX behavior subjects. + * + * A subject is a convenient way to create an observable when you want + * to fan out a single value to all subscribers. + * + * The BehaviorSubject will emit the most recently emitted value or error + * whenever a new observer subscribes to the subject. + * + * See http://reactivex.io/documentation/subject.html + */ +export class BehaviorSubject + implements Observable, ZenObservable.SubscriptionObserver { + private isClosed = false; + private currentValue: T; + private terminatingError?: Error; + + constructor(value: T) { + this.currentValue = value; + } + + private readonly observable = new ObservableImpl((subscriber) => { + if (this.isClosed) { + if (this.terminatingError) { + subscriber.error(this.terminatingError); + } else { + subscriber.complete(); + } + return () => {}; + } + + subscriber.next(this.currentValue); + + this.subscribers.add(subscriber); + return () => { + this.subscribers.delete(subscriber); + }; + }); + + private readonly subscribers = new Set< + ZenObservable.SubscriptionObserver + >(); + + get closed() { + return this.isClosed; + } + + next(value: T) { + if (this.isClosed) { + throw new Error('BehaviorSubject is closed'); + } + this.currentValue = value; + this.subscribers.forEach((subscriber) => subscriber.next(value)); + } + + error(error: Error) { + if (this.isClosed) { + throw new Error('BehaviorSubject is closed'); + } + this.isClosed = true; + this.terminatingError = error; + this.subscribers.forEach((subscriber) => subscriber.error(error)); + } + + complete() { + if (this.isClosed) { + throw new Error('BehaviorSubject is closed'); + } + this.isClosed = true; + this.subscribers.forEach((subscriber) => subscriber.complete()); + } + + subscribe(observer: ZenObservable.Observer): ZenObservable.Subscription; + subscribe( + onNext: (value: T) => void, + onError?: (error: any) => void, + onComplete?: () => void, + ): ZenObservable.Subscription; + subscribe( + onNext: ZenObservable.Observer | ((value: T) => void), + onError?: (error: any) => void, + onComplete?: () => void, + ): ZenObservable.Subscription { + const observer = + typeof onNext === 'function' + ? { + next: onNext, + error: onError, + complete: onComplete, + } + : onNext; + + return this.observable.subscribe(observer); + } +}