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

Ensure subscriptions always unsubscribe as required (#2380)

Browse Source 
* Ensure single-shot calls always unsubscribe

* memo/refCount cleanups

* Fix unsubscriptions

* Cleanup linting types

* CHANGELOG
This commit is contained in:
Jaco Greeff
2020-06-23 13:47:28 +02:00
committed by GitHub
parent daf3912920
commit 76da2638c1
7 changed files with 79 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/api-derive/src/util/memo.ts
+9 -9
View File
@@ -3,15 +3,11 @@
// of the Apache-2.0 license. See the LICENSE file for details.v
import createMemo from 'memoizee';
import { Observable, Observer } from 'rxjs';
import { Observable, Observer, TeardownLogic } from 'rxjs';
import { drr } from '@polkadot/rpc-core/rxjs';
type ObsFn <T> = (...params: any[]) => Observable<T>;
// Normalize via JSON.stringify, allow e.g. AccountId -> ss58
// eslint-disable-next-line @typescript-eslint/unbound-method
const normalizer = JSON.stringify;
// Wraps a derive, doing 2 things to optimize calls -
// 1. creates a memo of the inner fn -> Observable, removing when unsubscribed
// 2. wraps the observable in a drr() (which includes an unsub delay)
@@ -19,15 +15,19 @@ const normalizer = JSON.stringify;
export function memo <T> (inner: ObsFn<T>): ObsFn<T> {
const cached = createMemo(
(...params: any[]): Observable<T> =>
new Observable((observer: Observer<T>): VoidCallback => {
const sub = inner(...params).subscribe(observer);
new Observable((observer: Observer<T>): TeardownLogic => {
const subscription = inner(...params).subscribe(observer);
return (): void => {
cached.delete(...params);
sub.unsubscribe();
subscription.unsubscribe();
};
}).pipe(drr()),
{ normalizer }
{
// Normalize via JSON.stringify, allow e.g. AccountId -> ss58
// eslint-disable-next-line @typescript-eslint/unbound-method
normalizer: JSON.stringify
}
);
return cached;
packages/api/src/promise/Api.ts
+52 -33
View File
@@ -6,15 +6,15 @@ import { Callback, Codec } from '@polkadot/types/types';
import { ApiOptions, DecorateFn, DecorateMethodOptions, ObsInnerType, StorageEntryPromiseOverloads, UnsubscribePromise, VoidFn } from '../types';
import { EMPTY, Observable, Subscription } from 'rxjs';
import { catchError, first, tap } from 'rxjs/operators';
import { catchError, tap } from 'rxjs/operators';
import { isFunction, assert } from '@polkadot/util';
import ApiBase from '../base';
import Combinator, { CombinatorCallback, CombinatorFunction } from './Combinator';
interface Tracker {
interface Tracker<T> {
reject: (value: Error) => Observable<never>;
resolve: (value: () => void) => void;
resolve: (value: T) => void;
}
// extract the arguments and callback params from a value array possibly containing a callback
@@ -35,30 +35,63 @@ function extractArgs (args: unknown[], needsCallback: boolean): [unknown[], Call
}
// a Promise completion tracker, wrapping an isComplete variable that ensures the promise only resolves once
function promiseTracker (resolve: (value: VoidFn) => void, reject: (value: Error) => void): Tracker {
function promiseTracker<T> (resolve: (value: T) => void, reject: (value: Error) => void): Tracker<T> {
let isCompleted = false;
// eslint-disable-next-line @typescript-eslint/ban-types
const complete = (fn: Function, value: any): void => {
if (!isCompleted) {
isCompleted = true;
fn(value);
}
};
return {
reject: (error: Error): Observable<never> => {
complete(reject, error);
if (!isCompleted) {
isCompleted = true;
reject(error);
}
return EMPTY;
},
resolve: (value: any): void => {
complete(resolve, value);
resolve: (value: T): void => {
if (!isCompleted) {
isCompleted = true;
resolve(value);
}
}
};
}
// Decorate a call for a single-shot result - retrieve and then immediate unsubscribe
function decorateCall<Method extends DecorateFn<ObsInnerType<ReturnType<Method>>>> (method: Method, actualArgs: unknown[]): Promise<ObsInnerType<ReturnType<Method>>> {
return new Promise((resolve, reject): void => {
// single result tracker - either reject with Error or resolve with Codec result
const tracker = promiseTracker(resolve, reject);
// errors reject immediately, any result unsubscribes and resolves
const subscription: Subscription = method(...actualArgs).pipe(
catchError((error) => tracker.reject(error))
).subscribe((result): void => {
subscription.unsubscribe();
tracker.resolve(result);
});
});
}
// Decorate a subscription where we have a result callback specified
function decorateSubscribe<Method extends DecorateFn<ObsInnerType<ReturnType<Method>>>> (method: Method, actualArgs: unknown[], resultCb: Callback<Codec>): UnsubscribePromise {
return new Promise<VoidFn>((resolve, reject): void => {
// either reject with error or resolve with unsubscribe callback
const tracker = promiseTracker(resolve, reject);
// errors reject immediately, the first result resolves with an unsubscribe promise, all results via callback
const subscription: Subscription = method(...actualArgs).pipe(
catchError((error) => tracker.reject(error)),
tap(() => tracker.resolve(() => subscription.unsubscribe()))
).subscribe((result): void => {
// We use setImmediate here to ensure that the Promise above (tracker) has resolved, before returning
// the first result. By putting is back in the queue, the promise above is guarded for first execution
setImmediate(() => resultCb(result) as void);
});
});
}
/**
* @description Decorate method for ApiPromise, where the results are converted to the Promise equivalent
*/
@@ -68,23 +101,9 @@ export function decorateMethod<Method extends DecorateFn<ObsInnerType<ReturnType
return function (...args: unknown[]): Promise<ObsInnerType<ReturnType<Method>>> | UnsubscribePromise {
const [actualArgs, resultCb] = extractArgs(args, !!needsCallback);
if (!resultCb) {
return (options?.overrideNoSub || method)(...actualArgs).pipe(first()).toPromise() as Promise<ObsInnerType<ReturnType<Method>>>;
}
return new Promise<VoidFn>((resolve, reject): void => {
const tracker = promiseTracker(resolve, reject);
const subscription: Subscription = method(...actualArgs).pipe(
// if we find an error (invalid params, etc), reject the promise
catchError((error): Observable<never> => tracker.reject(error)),
// upon the first result, resolve with the unsub function
tap(() => tracker.resolve(() => subscription.unsubscribe()))
).subscribe((result): void => {
// We use setImmediate here to ensure that the Promise above (tracker) has resolved, before returning
// the first result. By putting is back in the queue, the promise above is guarder for first execution
setImmediate(() => resultCb(result) as void);
});
});
return resultCb
? decorateSubscribe(method, actualArgs, resultCb)
: decorateCall((options?.overrideNoSub as Method) || method, actualArgs);
} as StorageEntryPromiseOverloads;
}
packages/rpc-core/src/index.ts
+6 -6
View File
@@ -12,7 +12,7 @@ import { combineLatest, from, Observable, Observer, of, throwError } from 'rxjs'
import { catchError, map, publishReplay, refCount, switchMap } from 'rxjs/operators';
import jsonrpc from '@polkadot/types/interfaces/jsonrpc';
import { Option, StorageKey, Vec, createClass, createTypeUnsafe } from '@polkadot/types';
import { assert, hexToU8a, isFunction, isNull, isNumber, isUndefined, logger, u8aToU8a } from '@polkadot/util';
import { assert, hexToU8a, isFunction, isNull, isUndefined, logger, u8aToU8a } from '@polkadot/util';
import { drr } from './rxjs';
@@ -215,7 +215,7 @@ export default class Rpc implements RpcInterface {
}
// create a subscriptor, it subscribes once and resolves with the id as subscribe
private _createSubscriber ({ paramsJson, subName, subType, update }: { subType: string; subName: string; paramsJson: AnyJson[]; update: ProviderInterfaceCallback }, errorHandler: (error: Error) => void): Promise<number> {
private _createSubscriber ({ paramsJson, subName, subType, update }: { subType: string; subName: string; paramsJson: AnyJson[]; update: ProviderInterfaceCallback }, errorHandler: (error: Error) => void): Promise<number | string> {
return new Promise((resolve, reject): void => {
this.provider
.subscribe(subType, subName, paramsJson, update)
@@ -237,7 +237,7 @@ export default class Rpc implements RpcInterface {
const creator = (isRaw: boolean) => (...values: unknown[]): Observable<any> => {
return new Observable((observer: Observer<any>): VoidCallback => {
// Have at least an empty promise, as used in the unsubscribe
let subscriptionPromise: Promise<number | void> = Promise.resolve();
let subscriptionPromise: Promise<number | string | null> = Promise.resolve(null);
const errorHandler = (error: Error): void => {
logErrorMessage(method, def, error);
@@ -288,9 +288,9 @@ export default class Rpc implements RpcInterface {
// Unsubscribe from provider
subscriptionPromise
.then((subscriptionId): Promise<boolean> =>
isNumber(subscriptionId)
? this.provider.unsubscribe(subType, unsubName, subscriptionId)
: Promise.resolve(false)
isNull(subscriptionId)
? Promise.resolve(false)
: this.provider.unsubscribe(subType, unsubName, subscriptionId)
)
.catch((error: Error) => logErrorMessage(method, def, error));
};
packages/rpc-core/src/rxjs/refCountDelay.ts
+6 -4
View File
@@ -2,7 +2,7 @@
// This software may be modified and distributed under the terms
// of the Apache-2.0 license. See the LICENSE file for details.
import { asapScheduler, ConnectableObservable, MonoTypeOperatorFunction, Observable, Subscription } from 'rxjs';
import { asapScheduler, ConnectableObservable, MonoTypeOperatorFunction, Observable, Subscription, TeardownLogic } from 'rxjs';
const DELAY = 1750;
@@ -11,7 +11,7 @@ function refCountDelayInner <T> (source: Observable<T>): Observable<T> {
// state: 0 = disconnected, 1 = disconnecting, 2 = connecting, 3 = connected
let [state, refCount, connection, scheduler] = [0, 0, Subscription.EMPTY, Subscription.EMPTY];
return new Observable((ob) => {
return new Observable((ob): TeardownLogic => {
source.subscribe(ob);
if (refCount++ === 0) {
@@ -27,12 +27,14 @@ function refCountDelayInner <T> (source: Observable<T>): Observable<T> {
return (): void => {
if (--refCount === 0) {
if (state === 2) {
state = 0; scheduler.unsubscribe();
state = 0;
scheduler.unsubscribe();
} else {
// state === 3
state = 1;
scheduler = asapScheduler.schedule((): void => {
state = 0; connection.unsubscribe();
state = 0;
connection.unsubscribe();
}, DELAY);
}
}
packages/rpc-provider/src/types.ts
+2 -2
View File
@@ -49,6 +49,6 @@ export interface ProviderInterface {
isConnected (): boolean;
on (type: ProviderInterfaceEmitted, sub: ProviderInterfaceEmitCb): () => void;
send (method: string, params: any[]): Promise<any>;
subscribe (type: string, method: string, params: any[], cb: ProviderInterfaceCallback): Promise<number>;
unsubscribe (type: string, method: string, id: number): Promise<boolean>;
subscribe (type: string, method: string, params: any[], cb: ProviderInterfaceCallback): Promise<number | string>;
unsubscribe (type: string, method: string, id: number | string): Promise<boolean>;
}
packages/rpc-provider/src/ws/Provider.ts
+3 -3
View File
@@ -247,8 +247,8 @@ export default class WsProvider implements WSProviderInterface {
* })
* ```
*/
public async subscribe (type: string, method: string, params: any[], callback: ProviderInterfaceCallback): Promise<number> {
const id = await this.send(method, params, { callback, type }) as Promise<number>;
public async subscribe (type: string, method: string, params: any[], callback: ProviderInterfaceCallback): Promise<number | string> {
const id = await this.send(method, params, { callback, type }) as Promise<number | string>;
return id;
}
@@ -256,7 +256,7 @@ export default class WsProvider implements WSProviderInterface {
/**
* @summary Allows unsubscribing to subscriptions made with [[subscribe]].
*/
public async unsubscribe (type: string, method: string, id: number): Promise<boolean> {
public async unsubscribe (type: string, method: string, id: number | string): Promise<boolean> {
const subscription = `${type}::${id}`;
// FIXME This now could happen with re-subscriptions. The issue is that with a re-sub