diff --git a/CHANGELOG.md b/CHANGELOG.md index 0dd13099a3..a0bb6888d4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## 1.21.0-beta.x +- Fix subscriptions never unsubscribing after the id type swap in 1.18.1 - Promise API will default to using `getStorage` on non-subscription calls, reducing RPC overhead ## 1.20.1 Jun 22, 2020 diff --git a/packages/api-derive/src/util/memo.ts b/packages/api-derive/src/util/memo.ts index 73115b6627..aff6704f79 100644 --- a/packages/api-derive/src/util/memo.ts +++ b/packages/api-derive/src/util/memo.ts @@ -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 = (...params: any[]) => Observable; -// 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 (inner: ObsFn): ObsFn { const cached = createMemo( (...params: any[]): Observable => - new Observable((observer: Observer): VoidCallback => { - const sub = inner(...params).subscribe(observer); + new Observable((observer: Observer): 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; diff --git a/packages/api/src/promise/Api.ts b/packages/api/src/promise/Api.ts index 8ce87f5da4..f0b7a9a798 100644 --- a/packages/api/src/promise/Api.ts +++ b/packages/api/src/promise/Api.ts @@ -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 { reject: (value: Error) => Observable; - 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 (resolve: (value: T) => void, reject: (value: Error) => void): Tracker { 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 => { - 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: Method, actualArgs: unknown[]): Promise>> { + 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: Method, actualArgs: unknown[], resultCb: Callback): UnsubscribePromise { + return new Promise((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>> | UnsubscribePromise { const [actualArgs, resultCb] = extractArgs(args, !!needsCallback); - if (!resultCb) { - return (options?.overrideNoSub || method)(...actualArgs).pipe(first()).toPromise() as Promise>>; - } - - return new Promise((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 => 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; } diff --git a/packages/rpc-core/src/index.ts b/packages/rpc-core/src/index.ts index b916243165..10630d80d5 100644 --- a/packages/rpc-core/src/index.ts +++ b/packages/rpc-core/src/index.ts @@ -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 { + private _createSubscriber ({ paramsJson, subName, subType, update }: { subType: string; subName: string; paramsJson: AnyJson[]; update: ProviderInterfaceCallback }, errorHandler: (error: Error) => void): Promise { 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 => { return new Observable((observer: Observer): VoidCallback => { // Have at least an empty promise, as used in the unsubscribe - let subscriptionPromise: Promise = Promise.resolve(); + let subscriptionPromise: Promise = 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 => - 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)); }; diff --git a/packages/rpc-core/src/rxjs/refCountDelay.ts b/packages/rpc-core/src/rxjs/refCountDelay.ts index 7b8f5c69ca..c3c51cc82a 100644 --- a/packages/rpc-core/src/rxjs/refCountDelay.ts +++ b/packages/rpc-core/src/rxjs/refCountDelay.ts @@ -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 (source: Observable): Observable { // 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 (source: Observable): Observable { 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); } } diff --git a/packages/rpc-provider/src/types.ts b/packages/rpc-provider/src/types.ts index c1f198ace8..10b444cc2b 100644 --- a/packages/rpc-provider/src/types.ts +++ b/packages/rpc-provider/src/types.ts @@ -49,6 +49,6 @@ export interface ProviderInterface { isConnected (): boolean; on (type: ProviderInterfaceEmitted, sub: ProviderInterfaceEmitCb): () => void; send (method: string, params: any[]): Promise; - subscribe (type: string, method: string, params: any[], cb: ProviderInterfaceCallback): Promise; - unsubscribe (type: string, method: string, id: number): Promise; + subscribe (type: string, method: string, params: any[], cb: ProviderInterfaceCallback): Promise; + unsubscribe (type: string, method: string, id: number | string): Promise; } diff --git a/packages/rpc-provider/src/ws/Provider.ts b/packages/rpc-provider/src/ws/Provider.ts index 71b564ab2a..7b70ae982f 100644 --- a/packages/rpc-provider/src/ws/Provider.ts +++ b/packages/rpc-provider/src/ws/Provider.ts @@ -247,8 +247,8 @@ export default class WsProvider implements WSProviderInterface { * }) * ``` */ - public async subscribe (type: string, method: string, params: any[], callback: ProviderInterfaceCallback): Promise { - const id = await this.send(method, params, { callback, type }) as Promise; + public async subscribe (type: string, method: string, params: any[], callback: ProviderInterfaceCallback): Promise { + const id = await this.send(method, params, { callback, type }) as Promise; 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 { + public async unsubscribe (type: string, method: string, id: number | string): Promise { const subscription = `${type}::${id}`; // FIXME This now could happen with re-subscriptions. The issue is that with a re-sub