init

2019-04-11 13:42:06 +12:00
commit 4e30fab9d6
688 changed files with 51707 additions and 0 deletions
@@ -0,0 +1,88 @@
+// Copyright 2017-2019 @polkadot/rpc-rx 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 testingPairs from '@plugnet/keyring/testingPairs';
+import { RpcInterface$Section } from '@plugnet/rpc-core/types';
+import { AccountId } from '@plugnet/types';
+
+jest.mock('@polkadot/rpc-provider/ws', () => class {
+  isConnected = () => true;
+  on = () => true;
+  send = () => true;
+});
+
+import RpcRx from '.';
+
+describe('createCachedObservable', () => {
+  let api: RpcRx;
+  let creator: (...params: Array<any>) => Observable<any>;
+  const keyring = testingPairs();
+  let section: RpcInterface$Section;
+
+  beforeEach(() => {
+    api = new RpcRx();
+  });
+
+  beforeEach(() => {
+    // Create two methods in our section
+    const subMethod: any = jest.fn(() => Promise.resolve('subMethodResult'));
+    subMethod.unsubscribe = jest.fn(() => Promise.resolve(true));
+    const subMethod2: any = jest.fn(() => Promise.resolve('subMethod2Result'));
+    subMethod2.unsubscribe = jest.fn(() => Promise.resolve(true));
+
+    section = {
+      subMethod,
+      subMethod2
+    };
+
+    // @ts-ignore Private method
+    creator = api.createObservable('subMethod', section);
+  });
+
+  it('creates a single observable', () => {
+    creator(123).subscribe();
+
+    expect(
+      section.subMethod
+    ).toHaveBeenCalledWith(123, expect.anything());
+  });
+
+  it('creates a single observable (multiple calls)', () => {
+    const observable1 = creator(123);
+    const observable2 = creator(123);
+
+    expect(
+      observable2
+    ).toBe(observable1);
+  });
+
+  it('creates a single observable (multiple calls, different arguments that should be cached together)', () => {
+    const observable1 = creator(keyring.alice.address());
+    const observable2 = creator(new AccountId(keyring.alice.address()));
+
+    expect(
+      observable2
+    ).toBe(observable1);
+  });
+
+  it('creates multiple observables for different values', () => {
+    const observable1 = creator(123);
+    const observable2 = creator(456);
+
+    expect(
+      observable2
+    ).not.toBe(observable1);
+  });
+
+  it('creates different observables for different methods but same arguments', () => {
+    const observable1 = creator(123);
+    // @ts-ignore Private method
+    const observable2 = api.createObservable('subMethod2', section)(123);
+
+    expect(
+      observable2
+    ).not.toBe(observable1);
+  });
+});
@@ -0,0 +1,39 @@
+// Copyright 2017-2019 @polkadot/rpc-rx 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 { isFunction } from '@plugnet/util';
+
+jest.mock('@polkadot/rpc-provider/ws', () => class {
+  isConnected = () => true;
+  on = () => true;
+  send = () => true;
+});
+
+import RpcRx from '.';
+
+describe('RpcRx', () => {
+  let api: RpcRx;
+
+  beforeEach(() => {
+    api = new RpcRx();
+  });
+
+  it('creates an instance with all sections', () => {
+    expect(
+      Object
+        .keys(api)
+        .filter((key) =>
+          !key.startsWith('_')
+        )
+    ).toEqual([
+      'author', 'chain', 'state', 'system'
+    ]);
+  });
+
+  it('has isConnected', () => {
+    expect(
+      isFunction(api.isConnected)
+    ).toBe(true);
+  });
+});
@@ -0,0 +1,184 @@
+// Copyright 2017-2019 @polkadot/rpc-rx 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 { RpcInterface$Method, RpcInterface$Section } from '@plugnet/rpc-core/types';
+import { ProviderInterface } from '@plugnet/rpc-provider/types';
+import { RpcRxInterface, RpcRxInterface$Events, RpcRxInterface$Method, RpcRxInterface$Section } from './types';
+
+import EventEmitter from 'eventemitter3';
+import memoize, { Memoized } from 'memoizee';
+import { BehaviorSubject, Observable, from, Observer } from 'rxjs';
+import Rpc from '@plugnet/rpc-core';
+import { map, publishReplay, refCount } from 'rxjs/operators';
+import { isFunction, isUndefined } from '@plugnet/util';
+
+/**
+ * @name RpcRx
+ * @summary The RxJS API is a wrapper around the API.
+ * @description It allows wrapping API components with observables using RxJS.
+ *
+ * @example
+ * <BR>
+ *
+ * ```javascript
+ * import RpcRx from '@plugnet/rpc-rx';
+ * import WsProvider from '@plugnet/rpc-provider/ws';
+ *
+ * const provider = new WsProvider('http://127.0.0.1:9944');
+ * const api = new RpcRx(provider);
+ * ```
+ */
+export default class RpcRx implements RpcRxInterface {
+  private _api: Rpc;
+  private _eventemitter: EventEmitter;
+  private _isConnected: BehaviorSubject<boolean>;
+  readonly author: RpcRxInterface$Section;
+  readonly chain: RpcRxInterface$Section;
+  readonly state: RpcRxInterface$Section;
+  readonly system: RpcRxInterface$Section;
+
+  /**
+   * @param  {ProviderInterface} provider An API provider using HTTP or WebSocket
+   */
+  constructor (providerOrRpc?: Rpc | ProviderInterface) {
+    this._api = providerOrRpc instanceof Rpc
+      ? providerOrRpc
+      : new Rpc(providerOrRpc);
+    this._eventemitter = new EventEmitter();
+    this._isConnected = new BehaviorSubject(this._api._provider.isConnected());
+
+    this.initEmitters(this._api._provider);
+
+    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> {
+    return this._isConnected;
+  }
+
+  on (type: RpcRxInterface$Events, handler: (...args: Array<any>) => any): void {
+    this._eventemitter.on(type, handler);
+  }
+
+  protected emit (type: RpcRxInterface$Events, ...args: Array<any>): void {
+    this._eventemitter.emit(type, ...args);
+  }
+
+  private initEmitters (provider: ProviderInterface): void {
+    provider.on('connected', () => {
+      this._isConnected.next(true);
+
+      this.emit('connected');
+    });
+
+    provider.on('disconnected', () => {
+      this._isConnected.next(false);
+
+      this.emit('disconnected');
+    });
+  }
+
+  private createInterface (section: RpcInterface$Section): RpcRxInterface$Section {
+    return Object
+      .keys(section)
+      .filter((name) => !['subscribe', 'unsubscribe'].includes(name))
+      .reduce((observables, name) => {
+        observables[name] = this.createObservable(name, section);
+
+        return observables;
+      }, ({} as RpcRxInterface$Section));
+  }
+
+  private createObservable (name: string, section: RpcInterface$Section): RpcRxInterface$Method {
+    if (isFunction(section[name].unsubscribe)) {
+      const memoized: Memoized<RpcRxInterface$Method> = memoize(
+        (...params: Array<any>) => this.createReplay(name, params, section, memoized),
+        {
+          length: false,
+          // Normalize args so that different args that should be cached
+          // together are cached together.
+          // E.g.: `query.my.method('abc') === query.my.method(new AccountId('abc'));`
+          normalizer: (args) => {
+            // `args` is arguments object as accessible in memoized function
+            return JSON.stringify(args);
+          }
+        }
+      );
+
+      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]
+          .apply(null, params)
+          .catch((error: Error) => {
+            console.error(error);
+          })
+      );
+  }
+
+  private createReplay (name: string, params: Array<any>, section: RpcInterface$Section, memoized: Memoized<RpcRxInterface$Method>): Observable<any> {
+    return new Observable((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, params, memoized);
+    })
+      .pipe(
+        map((value) => {
+          if (value instanceof Error) {
+            throw value;
+          }
+
+          return value;
+        }),
+        publishReplay(1),
+        refCount()
+      );
+  }
+
+  private createReplayCallback (observer: Observer<any>) {
+    let cachedResult: any;
+
+    return (result: any) => {
+      if (isUndefined(cachedResult) || !Array.isArray(cachedResult) || !Array.isArray(result)
+        || result.length !== cachedResult.length) {
+        cachedResult = result;
+      } else {
+        cachedResult = cachedResult.map((cachedValue, index) =>
+          isUndefined(result[index])
+            ? cachedValue
+            : result[index]
+        );
+      }
+
+      observer.next(cachedResult);
+    };
+  }
+
+  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(() => {
+          memoized.delete(...params);
+        })
+        .catch((error) => {
+          console.error('Unsubscribe failed', error);
+        });
+    };
+  }
+}
@@ -0,0 +1,84 @@
+// Copyright 2017-2019 @polkadot/rpc-rx 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 { RpcInterface$Section } from '@plugnet/rpc-core/types';
+
+jest.mock('@polkadot/rpc-provider/ws', () => class {
+  isConnected = () => true;
+  on = () => true;
+  send = () => true;
+});
+
+import RpcRx from '.';
+
+describe('replay', () => {
+  const params = [123, false];
+  let api: RpcRx;
+  let section: RpcInterface$Section;
+  let observable: Observable<any>;
+  let update: any;
+
+  beforeEach(() => {
+    api = new RpcRx();
+  });
+
+  beforeEach(() => {
+    const subMethod: any = jest.fn((name, ...params) => {
+      update = params.pop();
+
+      return Promise.resolve(12345);
+    });
+
+    subMethod.unsubscribe = jest.fn(() => {
+      return Promise.resolve(true);
+    });
+
+    section = {
+      subMethod
+    };
+
+    // @ts-ignore
+    observable = api.createObservable('subMethod', section)(...params);
+  });
+
+  it('subscribes via the api section', (done) => {
+    observable.subscribe((value) => {
+      if (value) {
+        expect(
+          section.subMethod
+        ).toHaveBeenCalledWith(123, false, expect.anything());
+
+        done();
+      }
+    });
+
+    update('test');
+  });
+
+  it('returns the observable value', (done) => {
+    observable.subscribe((value) => {
+      if (value) {
+        expect(value).toEqual('test');
+        done();
+      }
+    });
+
+    update('test');
+  });
+
+  it('unsubscribes as required', (done) => {
+    section.subMethod.unsubscribe = async () => {
+      done();
+    };
+
+    let subscription = observable.subscribe((value) => {
+      if (value) {
+        subscription.unsubscribe();
+      }
+    });
+
+    update('test');
+  });
+});
@@ -0,0 +1,24 @@
+// Copyright 2017-2019 @polkadot/rpc-rx 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 { ProviderInterface$Emitted } from '@plugnet/rpc-provider/types';
+
+export type RpcRxInterface$Method = (...params: Array<any>) => Observable<any>;
+
+export type RpcRxInterface$Section = {
+  [index: string]: RpcRxInterface$Method
+};
+
+export type RpcRxInterface$Events = ProviderInterface$Emitted;
+
+export type RpcRxInterface = {
+  readonly author: RpcRxInterface$Section;
+  readonly chain: RpcRxInterface$Section;
+  readonly state: RpcRxInterface$Section;
+  readonly system: RpcRxInterface$Section;
+
+  isConnected: () => Observable<boolean>
+  on: (type: RpcRxInterface$Events, handler: (...args: Array<any>) => any) => void;
+};