Merge pull request #885 from spotify/rugvip/subjects

packages/core: add behavior and publish subjects to make it easier to produce observables

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<string | undefined>(undefined);
 
-  private readonly observable: Observable<string | undefined>;
-  private readonly subscribers = new Set<
-    ZenObservable.SubscriptionObserver<string | undefined>
-  >();
-
-  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<string | undefined> {
-    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);
   }
 }

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<ResultType> = {
   scopes: OAuthScopes;
@@ -44,16 +45,15 @@ export type OAuthPendingRequestsApi<ResultType> = {
 export class OAuthPendingRequests<ResultType>
   implements OAuthPendingRequestsApi<ResultType> {
   private requests: RequestQueueEntry<ResultType>[] = [];
-  private listeners: ZenObservable.SubscriptionObserver<
-    PendingRequest<ResultType>
-  >[] = [];
+  private subject = new BehaviorSubject<PendingRequest<ResultType>>(
+    this.getCurrentPending(),
+  );
 
   request(scopes: OAuthScopes): Promise<ResultType> {
     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<ResultType>
       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<PendingRequest<ResultType>> {
-    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<ResultType> {

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<PendingAuthRequest[]>;
-  private readonly observers = new Set<
-    ZenObservable.SubscriptionObserver<PendingAuthRequest[]>
-  >();
+  private readonly subject = new BehaviorSubject<PendingAuthRequest[]>([]);
   private currentRequests: PendingAuthRequest[] = [];
   private handlerCount = 0;
 
-  constructor() {
-    this.request$ = new Observable<PendingAuthRequest[]>((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<T>(options: AuthRequesterOptions<T>): AuthRequester<T> {
     const handler = new OAuthPendingRequests<T>();
 
@@ -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<PendingAuthRequest[]> {
-    return this.request$;
+    return this.subject;
   }
 
   async showLoginPopup(options: LoginPopupOptions): Promise<any> {

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';

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<number>();
+    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<number>();
+
+    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<number>();
+
+    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<number>(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);
+  });
+});

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<T>
+  implements Observable<T>, ZenObservable.SubscriptionObserver<T> {
+  private isClosed = false;
+  private terminatingError?: Error;
+
+  private readonly observable = new ObservableImpl<T>((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<T>
+  >();
+
+  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<T>): ZenObservable.Subscription;
+  subscribe(
+    onNext: (value: T) => void,
+    onError?: (error: any) => void,
+    onComplete?: () => void,
+  ): ZenObservable.Subscription;
+  subscribe(
+    onNext: ZenObservable.Observer<T> | ((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<T>
+  implements Observable<T>, ZenObservable.SubscriptionObserver<T> {
+  private isClosed = false;
+  private currentValue: T;
+  private terminatingError?: Error;
+
+  constructor(value: T) {
+    this.currentValue = value;
+  }
+
+  private readonly observable = new ObservableImpl<T>((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<T>
+  >();
+
+  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<T>): ZenObservable.Subscription;
+  subscribe(
+    onNext: (value: T) => void,
+    onError?: (error: any) => void,
+    onComplete?: () => void,
+  ): ZenObservable.Subscription;
+  subscribe(
+    onNext: ZenObservable.Observer<T> | ((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);
+  }
+}