quantus/api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use memoizee instead of custom cache (#643)

Browse Source 
* First run at using memoizee

* Make cached test pass

* Fix replay test

* Update e2e test

* Fix types

* Fix lint

* Simplify types

* Add comment
This commit is contained in:
Amaury Martiny
2019-01-28 18:46:46 +01:00
committed by GitHub
parent eccaa75192
commit e811d737f2
8 changed files with 133 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/rpc-rx/package.json
+2
View File
@@ -32,7 +32,9 @@
"@babel/runtime": "^7.2.0",
"@polkadot/rpc-core": "^0.41.7",
"@polkadot/rpc-provider": "^0.41.7",
"@types/memoizee": "^0.4.2",
"@types/rx": "^4.1.1",
"memoizee": "^0.4.14",
"rxjs": "^6.3.3"
}
}
packages/rpc-rx/src/cached.spec.js → packages/rpc-rx/src/cached.spec.ts
+14 -30
View File
@@ -2,46 +2,40 @@
// This software may be modified and distributed under the terms
// of the Apache-2.0 license. See the LICENSE file for details.
import { Observable } from 'rxjs';
import { RpcInterface$Section } from '@polkadot/rpc-core/types';
jest.mock('@polkadot/rpc-provider/ws', () => class {
isConnected = () => true;
on = () => true;
send = () => true;
});
const RpcRx = require('./index').default;
import RpcRx from './index';
describe('createCachedObservable', () => {
let api;
let creator;
let section;
let api: RpcRx;
let creator: (...params: Array<any>) => Observable<any>;
let section: RpcInterface$Section;
beforeEach(() => {
api = new RpcRx();
});
beforeEach(() => {
const subMethod = jest.fn((name, ...params) => {
return Promise.resolve(12345);
});
subMethod.unsubscribe = jest.fn(() => {
return Promise.resolve(true);
});
const resMethod = jest.fn((name, ...params) => {
return Promise.resolve(12345);
});
const subMethod: any = jest.fn(() => Promise.resolve(12345));
subMethod.unsubscribe = jest.fn(() => Promise.resolve(true));
section = {
resMethod,
subMethod
};
creator = api.createCachedObservable('test', 'subMethod', section);
// @ts-ignore
creator = api.createObservable('subMethod', section);
});
it('creates a single observable', () => {
creator(123).subscribe((value) => {});
creator(123).subscribe();
expect(
section.subMethod
@@ -50,27 +44,17 @@ describe('createCachedObservable', () => {
it('creates a single observable (multiple calls)', () => {
const observable1 = creator(123);
observable1.subscribe((value) => {});
const observable2 = creator(123);
observable2.subscribe((value) => {});
expect(
observable2
).toEqual(observable1);
).toBe(observable1);
});
it('creates multiple observers for different values', () => {
it('creates multiple observables for different values', () => {
const observable1 = creator(123);
observable1.subscribe((value) => {});
const observable2 = creator(456);
observable2.subscribe((value) => {});
expect(
observable2
).not.toEqual(observable1);
packages/rpc-rx/src/index.spec.js → packages/rpc-rx/src/index.spec.ts
+2 -2
View File
@@ -10,10 +10,10 @@ jest.mock('@polkadot/rpc-provider/ws', () => class {
send = () => true;
});
const RpcRx = require('./index').default;
import RpcRx from './index';
describe('RpcRx', () => {
let api;
let api: RpcRx;
beforeEach(() => {
api = new RpcRx();
packages/rpc-rx/src/index.ts
+27 -42
View File
@@ -4,20 +4,15 @@
import { RpcInterface$Method, RpcInterface$Section } from '@polkadot/rpc-core/types';
import { ProviderInterface } from '@polkadot/rpc-provider/types';
import { RpcRxInterface, RpcRxInterface$Events, RpcRxInterface$Section } from './types';
import { RpcRxInterface, RpcRxInterface$Events, RpcRxInterface$Method, RpcRxInterface$Section } from './types';
import EventEmitter from 'eventemitter3';
import { BehaviorSubject, Observable, Subscriber, from } from 'rxjs';
import { map, publishReplay, refCount } from 'rxjs/operators';
import memoize, { Memoized } from 'memoizee';
import { BehaviorSubject, Observable, from, Observer } from 'rxjs';
import Rpc from '@polkadot/rpc-core/index';
import { map, publishReplay, refCount } from 'rxjs/operators';
import { isFunction, isUndefined } from '@polkadot/util';
type CachedMap = {
[index: string]: {
[index: string]: Observable<any>
}
};
/**
* @name RpcRx
* @summary The RxJS API is a wrapper around the API.
@@ -36,7 +31,6 @@ type CachedMap = {
*/
export default class RpcRx implements RpcRxInterface {
private _api: Rpc;
private _cacheMap: CachedMap;
private _eventemitter: EventEmitter;
private _isConnected: BehaviorSubject<boolean>;
readonly author: RpcRxInterface$Section;
@@ -51,16 +45,15 @@ export default class RpcRx implements RpcRxInterface {
this._api = providerOrRpc instanceof Rpc
? providerOrRpc
: new Rpc(providerOrRpc);
this._cacheMap = {};
this._eventemitter = new EventEmitter();
this._isConnected = new BehaviorSubject(this._api._provider.isConnected());
this.initEmitters(this._api._provider);
this.author = this.createInterface('author', this._api.author);
this.chain = this.createInterface('chain', this._api.chain);
this.state = this.createInterface('state', this._api.state);
this.system = this.createInterface('system', this._api.system);
this.author = this.createInterface(this._api.author);
this.chain = this.createInterface(this._api.chain);
this.state = this.createInterface(this._api.state);
this.system = this.createInterface(this._api.system);
}
isConnected (): BehaviorSubject<boolean> {
@@ -89,22 +82,30 @@ export default class RpcRx implements RpcRxInterface {
});
}
private createInterface (sectionName: string, section: RpcInterface$Section): RpcRxInterface$Section {
private createInterface (section: RpcInterface$Section): RpcRxInterface$Section {
return Object
.keys(section)
.filter((name) => !['subscribe', 'unsubscribe'].includes(name))
.reduce((observables, name) => {
observables[name] = this.createObservable(`${sectionName}_${name}`, name, section);
observables[name] = this.createObservable(name, section);
return observables;
}, ({} as RpcRxInterface$Section));
}
private createObservable (subName: string, name: string, section: RpcInterface$Section): (...params: Array<any>) => Observable<any> {
private createObservable (name: string, section: RpcInterface$Section): RpcRxInterface$Method {
if (isFunction(section[name].unsubscribe)) {
return this.createCachedObservable(subName, name, section);
const memoized: Memoized<RpcRxInterface$Method> = memoize(
(...params: Array<any>) => this.createReplay(name, params, section, memoized),
{ length: false }
);
return memoized as unknown as RpcRxInterface$Method;
}
// We voluntarily don't cache the "one-shot" RPC calls. For example,
// `getStorage('123')` returns the current value, but this value can change
// over time, so we wouldn't want to cache the Observable.
return (...params: Array<any>): Observable<any> =>
from(
section[name]
@@ -115,32 +116,16 @@ export default class RpcRx implements RpcRxInterface {
);
}
private createCachedObservable (subName: string, name: string, section: RpcInterface$Section): (...params: Array<any>) => Observable<any> {
if (!this._cacheMap[subName]) {
this._cacheMap[subName] = {};
}
return (...params: Array<any>): Observable<any> => {
const paramStr = JSON.stringify(params);
if (!this._cacheMap[subName][paramStr]) {
this._cacheMap[subName][paramStr] = this.createReplay(name, params, section, subName, paramStr);
}
return this._cacheMap[subName][paramStr];
};
}
private createReplay (name: string, params: Array<any>, section: RpcInterface$Section, subName: string, paramStr: string): Observable<any> {
private createReplay (name: string, params: Array<any>, section: RpcInterface$Section, memoized: Memoized<RpcRxInterface$Method>): Observable<any> {
return Observable
.create((observer: Subscriber<any>): Function => {
.create((observer: Observer<any>): Function => {
const fn = section[name];
const callback = this.createReplayCallback(observer);
const subscribe = fn(...params, callback).catch((error) =>
observer.next(error)
);
return this.createReplayUnsub(fn, subscribe, subName, paramStr);
return this.createReplayUnsub(fn, subscribe, params, memoized);
})
.pipe(
map((value) => {
@@ -155,12 +140,12 @@ export default class RpcRx implements RpcRxInterface {
);
}
private createReplayCallback (observer: Subscriber<any>) {
private createReplayCallback (observer: Observer<any>) {
let cachedResult: any;
return (result: any) => {
if (isUndefined(cachedResult) || !Array.isArray(cachedResult) || !Array.isArray(result)
|| result.length !== cachedResult.length) {
|| result.length !== cachedResult.length) {
cachedResult = result;
} else {
cachedResult = cachedResult.map((cachedValue, index) =>
@@ -174,14 +159,14 @@ export default class RpcRx implements RpcRxInterface {
};
}
private createReplayUnsub (fn: RpcInterface$Method, subscribe: Promise<number>, subName: string, paramStr: string): () => void {
private createReplayUnsub (fn: RpcInterface$Method, subscribe: Promise<number>, params: Array<any>, memoized: Memoized<RpcRxInterface$Method>): () => void {
return (): void => {
subscribe
.then((subscriptionId: number) =>
fn.unsubscribe(subscriptionId)
)
.then(() => {
delete this._cacheMap[subName][paramStr];
memoized.delete(...params);
})
.catch((error) => {
console.error('Unsubscribe failed', error);
packages/rpc-rx/src/replay.js → packages/rpc-rx/src/replay.spec.ts
+11 -7
View File
@@ -2,27 +2,30 @@
// This software may be modified and distributed under the terms
// of the Apache-2.0 license. See the LICENSE file for details.
import { Observable } from 'rxjs';
import { RpcInterface$Section } from '@polkadot/rpc-core/types';
jest.mock('@polkadot/rpc-provider/ws', () => class {
isConnected = () => true;
on = () => true;
send = () => true;
});
const RpcRx = require('./index').default;
import RpcRx from './index';
describe('replay', () => {
const params = [123, false];
let api;
let update;
let section;
let observable;
let api: RpcRx;
let section: RpcInterface$Section;
let observable: Observable<any>;
let update: any;
beforeEach(() => {
api = new RpcRx();
});
beforeEach(() => {
const subMethod = jest.fn((name, ...params) => {
const subMethod: any = jest.fn((name, ...params) => {
update = params.pop();
return Promise.resolve(12345);
@@ -36,7 +39,8 @@ describe('replay', () => {
subMethod
};
observable = api.createReplay('subMethod', params, section);
// @ts-ignore
observable = api.createObservable('subMethod', section)(...params);
});
it('subscribes via the api section', (done) => {
packages/rpc-rx/src/types.ts
+2 -2
View File
@@ -2,10 +2,10 @@
// This software may be modified and distributed under the terms
// of the Apache-2.0 license. See the LICENSE file for details.
import { ReplaySubject, Observable } from 'rxjs';
import { Observable } from 'rxjs';
import { ProviderInterface$Emitted } from '@polkadot/rpc-provider/types';
export type RpcRxInterface$Method = (...params: Array<any>) => Observable<any> | ReplaySubject<any>;
export type RpcRxInterface$Method = (...params: Array<any>) => Observable<any>;
export type RpcRxInterface$Section = {
[index: string]: RpcRxInterface$Method