From 1d3eecdb775221735ab31fbfe111a7a716426f94 Mon Sep 17 00:00:00 2001 From: Jaco Greeff Date: Wed, 7 Nov 2018 12:43:53 +0100 Subject: [PATCH] Allow replaySubject to unsubscribe (#373) * Allow replaySubject to unsubscribe * Cleanup types, cache observable is not a subject, * ReplaySubject now unused * Cleanup CodeClimate * Cleanup helpers --- packages/rpc-rx/src/cached.spec.js | 13 ---- packages/rpc-rx/src/index.ts | 60 ++++++++++--------- .../rpc-rx/src/{subject.spec.js => replay.js} | 27 +++++---- 3 files changed, 49 insertions(+), 51 deletions(-) rename packages/rpc-rx/src/{subject.spec.js => replay.js} (75%) diff --git a/packages/rpc-rx/src/cached.spec.js b/packages/rpc-rx/src/cached.spec.js index fc9d3753b9..a7cb69759c 100644 --- a/packages/rpc-rx/src/cached.spec.js +++ b/packages/rpc-rx/src/cached.spec.js @@ -75,17 +75,4 @@ describe('createCachedObservable', () => { observable2 ).not.toEqual(observable1); }); - - it('functions as an subject', (done) => { - const subject = creator(123); - - subject.subscribe((value) => { - if (value) { - expect(value).toEqual('test'); - done(); - } - }); - - subject.next('test'); - }); }); diff --git a/packages/rpc-rx/src/index.ts b/packages/rpc-rx/src/index.ts index f2cea00d75..c7048b599d 100644 --- a/packages/rpc-rx/src/index.ts +++ b/packages/rpc-rx/src/index.ts @@ -2,18 +2,19 @@ // This software may be modified and distributed under the terms // of the ISC license. See the LICENSE file for details. -import { RpcInterface$Section } from '@polkadot/rpc-core/types'; +import { RpcInterface$Section, RpcInterface$Section$Method } from '@polkadot/rpc-core/types'; import { ProviderInterface } from '@polkadot/rpc-provider/types'; import { RpcRxInterface, RpcRxInterface$Events, RpcRxInterface$Section } from './types'; import EventEmitter from 'eventemitter3'; -import { BehaviorSubject, ReplaySubject, Observable, Subscriber, from } from 'rxjs'; +import { BehaviorSubject, Observable, Subscriber, from } from 'rxjs'; +import { publishReplay, refCount } from 'rxjs/operators'; import Rpc from '@polkadot/rpc-core/index'; import { isFunction, isUndefined } from '@polkadot/util'; type CachedMap = { [index: string]: { - [index: string]: ReplaySubject + [index: string]: Observable } }; @@ -99,7 +100,7 @@ export default class RpcRx implements RpcRxInterface { }, ({} as RpcRxInterface$Section)); } - private createObservable (subName: string, name: string, section: RpcInterface$Section): (...params: Array) => Observable | ReplaySubject { + private createObservable (subName: string, name: string, section: RpcInterface$Section): (...params: Array) => Observable { if (isFunction(section[name].unsubscribe)) { return this.createCachedObservable(subName, name, section); } @@ -114,43 +115,30 @@ export default class RpcRx implements RpcRxInterface { ); } - private createCachedObservable (subName: string, name: string, section: RpcInterface$Section): (...params: Array) => ReplaySubject { + private createCachedObservable (subName: string, name: string, section: RpcInterface$Section): (...params: Array) => Observable { if (!this._cacheMap[subName]) { this._cacheMap[subName] = {}; } - return (...params: Array): ReplaySubject => { + return (...params: Array): Observable => { const paramStr = JSON.stringify(params); if (!this._cacheMap[subName][paramStr]) { - this._cacheMap[subName][paramStr] = this.createSubject(name, params, section); + this._cacheMap[subName][paramStr] = this.createReplay(name, params, section, subName, paramStr); } return this._cacheMap[subName][paramStr]; }; } - private createSubject (name: string, params: Array, section: RpcInterface$Section, unsubCallback?: () => void): ReplaySubject { - const subject = new ReplaySubject(1); - - Observable + private createReplay (name: string, params: Array, section: RpcInterface$Section, subName: string, paramStr: string): Observable { + return Observable .create((observer: Subscriber): Function => { try { const fn = section[name]; - const subscribe = fn(...params, this.createSubjectCallback(observer)); + const subscribe = fn(...params, this.createReplayCallback(observer)); - return (): void => { - subscribe - .then((subscriptionId: number) => - fn.unsubscribe(subscriptionId) - ) - .then(() => - isFunction(unsubCallback) && unsubCallback() - ) - .catch((error) => { - console.error('Unsubscribe failed', error); - }); - }; + return this.createReplayUnsub(fn, subscribe, subName, paramStr); } catch (error) { console.error(error); @@ -159,12 +147,13 @@ export default class RpcRx implements RpcRxInterface { }; } }) - .subscribe(subject); - - return subject; + .pipe( + publishReplay(1), + refCount() + ); } - private createSubjectCallback (observer: Subscriber) { + private createReplayCallback (observer: Subscriber) { let cachedResult: any; return (result: any) => { @@ -182,4 +171,19 @@ export default class RpcRx implements RpcRxInterface { observer.next(cachedResult); }; } + + private createReplayUnsub (fn: RpcInterface$Section$Method, subscribe: Promise, subName: string, paramStr: string): () => void { + return (): void => { + subscribe + .then((subscriptionId: number) => + fn.unsubscribe(subscriptionId) + ) + .then(() => { + delete this._cacheMap[subName][paramStr]; + }) + .catch((error) => { + console.error('Unsubscribe failed', error); + }); + }; + } } diff --git a/packages/rpc-rx/src/subject.spec.js b/packages/rpc-rx/src/replay.js similarity index 75% rename from packages/rpc-rx/src/subject.spec.js rename to packages/rpc-rx/src/replay.js index b7a36671ca..1f4629b46f 100644 --- a/packages/rpc-rx/src/subject.spec.js +++ b/packages/rpc-rx/src/replay.js @@ -10,12 +10,11 @@ jest.mock('@polkadot/rpc-provider/ws', () => class { const RpcRx = require('./index').default; -describe('subject', () => { +describe('replay', () => { const params = [123, false]; let api; let update; let section; - let subscription; let observable; beforeEach(() => { @@ -37,13 +36,7 @@ describe('subject', () => { subMethod }; - observable = api.createSubject('subMethod', params, section); - }); - - afterEach(() => { - if (subscription) { - subscription.unsubscribe(); - } + observable = api.createReplay('subMethod', params, section); }); it('subscribes via the api section', (done) => { @@ -61,7 +54,7 @@ describe('subject', () => { }); it('returns the observable value', (done) => { - subscription = observable.subscribe((value) => { + observable.subscribe((value) => { if (value) { expect(value).toEqual('test'); done(); @@ -70,4 +63,18 @@ describe('subject', () => { update('test'); }); + + it('unsubscribes as required', (done) => { + section.subMethod.unsubscribe = async () => { + done(); + }; + + let subscription = observable.subscribe((value) => { + if (value) { + subscription.unsubscribe(); + } + }); + + update('test'); + }); });