Sub/unsub cleanups (#1535)

* Sub/unsub cleanups * drr full idAndIndex * Memo on nick call * refCountDelay * Remove scheduler param, always asap * Flatten * Test cleanups * Make tests pass * Remove logs * Drop nick memo * sourceConnection.unsubscribe * Small cleanups
2019-11-01 14:17:47 +01:00
parent f41ad2eee7
commit dc105e6b31
42 changed files with 209 additions and 110 deletions
@@ -44,22 +44,41 @@ describe('Cached Observables', (): void => {
    expect(observable2).not.toBe(observable1);
  });

-  it('clears cache if there are no more subscribers', (): void => {
+  it('subscribes to the same one if within the period (unbsub delay)', (done): void => {
+    const observable1 = rpc.chain.subscribeNewHeads();
+    const sub1 = observable1.subscribe();
+
+    sub1.unsubscribe();
+
+    setTimeout((): void => {
+      const observable2 = rpc.chain.subscribeNewHeads();
+      const sub2 = observable2.subscribe();
+
+      expect(observable1).toBe(observable2);
+
+      sub2.unsubscribe();
+      done();
+    }, 500);
+  });
+
+  it('clears cache if there are no more subscribers', (done): void => {
    const observable1 = rpc.chain.subscribeNewHeads();
    const observable2 = rpc.chain.subscribeNewHeads();
-
-    expect(observable1).toBe(observable2);
-
    const sub1 = observable1.subscribe();
    const sub2 = observable2.subscribe();

+    expect(observable1).toBe(observable2);
+
    sub1.unsubscribe();
    sub2.unsubscribe();

-    // No more subscribers, now create a new observable
-    const observable3 = rpc.chain.subscribeNewHeads();
+    setTimeout((): void => {
+      // No more subscribers, now create a new observable
+      const observable3 = rpc.chain.subscribeNewHeads();

-    expect(observable3).not.toBe(observable1);
+      expect(observable3).not.toBe(observable1);
+      done();
+    }, 3500);
  });

  it('creates different observables for different methods but same arguments', (): void => {
@@ -10,7 +10,7 @@ import { RpcInterfaceMethod, UserRpc } from './types';

 import memoizee from 'memoizee';
 import { combineLatest, from, Observable, Observer, of, throwError } from 'rxjs';
-import { catchError, distinctUntilChanged, map, publishReplay, refCount, switchMap } from 'rxjs/operators';
+import { catchError, map, publishReplay, refCount, switchMap } from 'rxjs/operators';
 import jsonrpc from '@polkadot/jsonrpc';
 import jsonrpcMethod from '@polkadot/jsonrpc/create/method';
 import jsonrpcParam from '@polkadot/jsonrpc/create/param';
@@ -18,6 +18,8 @@ import { Option, StorageKey, Vec, createClass } from '@polkadot/types';
 import { createTypeUnsafe } from '@polkadot/types/codec';
 import { assert, isFunction, isNull, isNumber, logger, u8aToU8a } from '@polkadot/util';

+import { drr } from './rxjs';
+
 type UserRpcConverted = Record<string, Record<string, RpcMethod>>;

 const l = logger('rpc-core');
@@ -223,6 +225,19 @@ export default class Rpc implements RpcInterface {
    return call as RpcInterfaceMethod;
  }

+  // create a subscriptor, it subscribes once and resolves with the id as subscribe
+  private createSubscriber ({ subType, subName, paramsJson, update }: { subType: string; subName: string; paramsJson: AnyJson[]; update: (error?: Error, result?: any) => void }, errorHandler: (error: Error) => void): Promise<number> {
+    return new Promise((resolve, reject): void => {
+      this.provider
+        .subscribe(subType, subName, paramsJson, update)
+        .then(resolve)
+        .catch((error): void => {
+          errorHandler(error);
+          reject(error);
+        });
+    });
+  }
+
  private createMethodSubscribe (method: RpcMethod): RpcInterfaceMethod {
    const [updateType, subMethod, unsubMethod] = method.pubsub;
    const subName = `${method.section}_${subMethod}`;
@@ -253,9 +268,7 @@ export default class Rpc implements RpcInterface {
            observer.next(this.formatOutput(method, params, result));
          };

-          subscriptionPromise = this.provider
-            .subscribe(subType, subName, paramsJson, update)
-            .catch(errorHandler);
+          subscriptionPromise = this.createSubscriber({ subType, subName, paramsJson, update }, errorHandler);
        } catch (error) {
          errorHandler(error);
        }
@@ -281,18 +294,9 @@ export default class Rpc implements RpcInterface {
                ? this.provider.unsubscribe(subType, unsubName, subscriptionId)
                : Promise.resolve(false)
            )
-            .catch((error: Error): void =>
-              l.error(this.createErrorMessage(method, error))
-            );
+            .catch((error: Error): void => l.error(this.createErrorMessage(method, error)));
        };
-      }).pipe(
-        // Duplicated in api-derive/util/drr
-        distinctUntilChanged((a: any, b: any): boolean =>
-          JSON.stringify({ value: a }) === JSON.stringify({ value: b })
-        ),
-        publishReplay(1),
-        refCount()
-      );
+      }).pipe(drr());
    };

    const memoized = memoizee(call, {
@@ -57,12 +57,12 @@ describe('replay', (): void => {
    const subscription = rpc.chain.subscribeNewHeads().subscribe((): void => {
      subscription.unsubscribe();

-      // There's a promise inside .unsubscribe(), wait a bit
+      // There's a promise inside .unsubscribe(), wait a bit (> 2s)
      setTimeout((): void => {
        // eslint-disable-next-line @typescript-eslint/unbound-method
        expect(rpc.provider.unsubscribe).toHaveBeenCalled();
        done();
-      }, 200);
+      }, 3500);
    });
  });
 });
@@ -0,0 +1,32 @@
+// Copyright 2017-2019 @polkadot/api-derive authors & contributors
+// This software may be modified and distributed under the terms
+// of the Apache-2.0 license. See the LICENSE file for details.
+
+import { of, timer } from 'rxjs';
+
+import { drr } from './drr';
+
+describe('drr', (): void => {
+  it('should not fire twice the same value', (done): void => {
+    let count = 0;
+    of(1, 1).pipe(drr()).subscribe((): void => { ++count; });
+
+    setTimeout((): void => {
+      expect(count).toBe(1);
+      done();
+    }, 50);
+  });
+
+  it('should be a ReplaySubject(1)', (done): void => {
+    const obs = timer(0, 100).pipe(drr()); // Starts at 0, increments every 100ms
+    obs.subscribe(); // Fire the observable
+
+    // Subscribe another time after some time, i.e. after the observable has fired
+    setTimeout((): void => {
+      obs.subscribe((value): void => {
+        expect(value > 1).toBe(true);
+        done();
+      });
+    }, 500);
+  });
+});
@@ -0,0 +1,32 @@
+// Copyright 2017-2019 @polkadot/rpc-core authors & contributors
+// 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 { catchError, distinctUntilChanged, publishReplay } from 'rxjs/operators';
+import { logger } from '@polkadot/util';
+
+import { refCountDelay } from './refCountDelay';
+
+type DrrResult = <T> (source$: Observable<T>) => Observable<T>;
+
+const l = logger('drr');
+
+/**
+ * Shorthand for distinctUntilChanged(), publishReplay(1) and refCount().
+ *
+ * @ignore
+ */
+export const drr = (): DrrResult => <T> (source$: Observable<T>): Observable<T> =>
+  source$.pipe(
+    catchError((error): Observable<never> => {
+      l.error(error);
+
+      throw error;
+    }),
+    distinctUntilChanged((a: any, b: any): boolean =>
+      JSON.stringify({ t: a }) === JSON.stringify({ t: b })
+    ),
+    publishReplay(1),
+    refCountDelay()
+  );
@@ -0,0 +1,6 @@
+// Copyright 2017-2019 @polkadot/rpc-core authors & contributors
+// This software may be modified and distributed under the terms
+// of the Apache-2.0 license. See the LICENSE file for details.
+
+export * from './drr';
+export * from './refCountDelay';
@@ -0,0 +1,55 @@
+// Copyright 2017-2019 @polkadot/rpc-core authors & contributors
+// This software may be modified and distributed under the terms
+// of the Apache-2.0 license. See the LICENSE file for details.
+
+// From https://medium.com/@volkeron/rxjs-unsubscribe-delay-218a9ab2672e (adapted for rxjs 6)
+// v6 inspiration from https://github.com/cartant/rxjs-etc/blob/7d7cd21233f67c364fd246ef72077bbe494dd5c2/source/operators/refCountDelay.ts
+
+import { asapScheduler, ConnectableObservable, MonoTypeOperatorFunction, NEVER, Observable, Subject, Subscription, timer, using } from 'rxjs';
+import { scan, switchMap, tap } from 'rxjs/operators';
+
+const DELAY = 1750;
+
+// subscribe to a source, tracker and increment the counter
+function subscribe <T> (source: Observable<T>, counter: Subject<number>, tracker: Observable<number>): [Subscription, Subscription] {
+  const sourceConn = (source as ConnectableObservable<T>).connect();
+  const trackerConn = tracker.subscribe();
+
+  counter.next(1);
+
+  return [sourceConn, trackerConn];
+}
+
+// unsubscribe from both the tracker and the source
+function unsubscribe (sourceConn: Subscription | undefined, trackerConn: Subscription | undefined): void {
+  sourceConn && sourceConn.unsubscribe();
+  trackerConn && trackerConn.unsubscribe();
+}
+
+// keep track of references, but only unsubscribe after the configured delay
+export function refCountDelay <T> (): MonoTypeOperatorFunction<T> {
+  return (source: Observable<T>): Observable<T> => {
+    let sourceConn: Subscription | undefined;
+    let trackerConn: Subscription | undefined;
+    const counter = new Subject<number>();
+    const tracker = counter.pipe(
+      scan((total, change) => change + total, 0),
+      switchMap((count) =>
+        count === 0 // when we have a zero count, schedule a unsubscribe
+          ? timer(DELAY, asapScheduler).pipe(tap((): void => unsubscribe(sourceConn, trackerConn)))
+          : NEVER
+      )
+    );
+
+    return using(
+      () => {
+        [sourceConn, trackerConn] = subscribe(source, counter, tracker);
+
+        return {
+          unsubscribe: (): void => counter.next(-1)
+        };
+      },
+      (): Observable<T> => source
+    );
+  };
+}