packages/core: add basic implementations of bahavior and publish RX subjects

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