From 06bf6fa6aa81e9c024092984ce5b49a7c5de7df6 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Fri, 15 May 2020 13:36:49 +0200 Subject: [PATCH] packages/core: add basic implementations of bahavior and publish RX subjects --- .../src/api/apis/implementations/lib/index.ts | 17 ++ .../apis/implementations/lib/subjects.test.ts | 178 +++++++++++++++ .../api/apis/implementations/lib/subjects.ts | 202 ++++++++++++++++++ 3 files changed, 397 insertions(+) create mode 100644 packages/core/src/api/apis/implementations/lib/index.ts create mode 100644 packages/core/src/api/apis/implementations/lib/subjects.test.ts create mode 100644 packages/core/src/api/apis/implementations/lib/subjects.ts 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); + } +}