Make frequency observables work

packages/light.js/src/api.ts

@@ -6,9 +6,14 @@
 import * as Api from '@parity/api';
 import * as debug from 'debug';
 import * as EventEmitter from 'eventemitter3';
+import * as memoizee from 'memoizee';
 
 let api: any; // TODO @parity/api
 
+export const createApiFromProvider = memoizee(
+  (provider?: any) => new Api(provider)
+);
+
 /**
  * Use this null Api provider if the Api hasn't been set by the end user yet.
  *
@@ -24,27 +29,13 @@ export class NullProvider extends EventEmitter {
   }
 }
 
-/**
- * Sets a new Api object.
- *
- * @param newApi - An Api object.
- */
-export const setApi = (newApi: any) => {
-  api = newApi;
-  if (!api.isPubSub) {
-    console.warn(
-      `Current provider does not support pubsub. @parity/light.js will poll every second to listen to changes.`
-    );
-  }
-};
-
 /**
  * Sets a new Ethereum provider object.
  *
  * @param provider - An Ethereum provider object.
  */
-export const setProvider = (provider: any) => {
-  api = new Api(provider);
+export const setProvider = (provider?: any) => {
+  api = createApiFromProvider(provider);
   if (!api.isPubSub) {
     console.warn(
       `Current provider does not support pubsub. @parity/light.js will poll every second to listen to changes.`
@@ -61,7 +52,7 @@ export const setProvider = (provider: any) => {
  */
 export const getApi = () => {
   if (!api) {
-    api = new Api(new NullProvider());
+    api = createApiFromProvider(new NullProvider());
   }
   return api;
 };

packages/light.js/src/frequency/accounts.ts

@@ -4,23 +4,24 @@
 // SPDX-License-Identifier: MIT
 
 import { AccountsInfo, Address, FrequencyObservable } from '../types';
-import api from '../api';
 import createPubsubObservable from './utils/createPubsubObservable';
 
 /**
  * Observable that emits each time the default account changes
+ *
+ * @param provider - The provider object with which to create this {@link FrequencyObservable}.
  */
-export const onAccountsChanged$ = (() =>
-  createPubsubObservable('eth_accounts', api)) as FrequencyObservable<
-  Address[]
->;
+export const onAccountsChanged$: FrequencyObservable<Address[]> = (
+  provider?: any
+) => createPubsubObservable('eth_accounts', provider);
 onAccountsChanged$.metadata = { name: 'onAccountsChanged$' };
 
 /**
  * Observable that emits each time the default account changes
+ *
+ * @param provider - The provider object with which to create this {@link FrequencyObservable}.
  */
-export const onAccountsInfoChanged$ = (() =>
-  createPubsubObservable('parity_accountsInfo', api)) as FrequencyObservable<
-  AccountsInfo
->;
+export const onAccountsInfoChanged$: FrequencyObservable<AccountsInfo> = (
+  provider?: any
+) => createPubsubObservable('parity_accountsInfo', provider);
 onAccountsInfoChanged$.metadata = { name: 'onAccountsInfoChanged$' };

packages/light.js/src/frequency/blocks.ts

@@ -6,33 +6,40 @@
 import BigNumber from 'bignumber.js';
 import { filter } from 'rxjs/operators';
 
-import api from '../api';
 import createPubsubObservable from './utils/createPubsubObservable';
 import { FrequencyObservable } from '../types';
 
 /**
  * Observable that emits on every new block.
+ *
+ * @param provider - The provider object with which to create this {@link FrequencyObservable}.
  */
-export const onEveryBlock$ = (() =>
-  createPubsubObservable('eth_blockNumber', api)) as FrequencyObservable<
-  BigNumber
->;
+export const onEveryBlock$: FrequencyObservable<BigNumber> = (provider?: any) =>
+  createPubsubObservable('eth_blockNumber', provider);
 onEveryBlock$.metadata = { name: 'onEveryBlock$' };
 
 /**
  * Observable that emits on every 2nd block.
+ *
+ * @param provider - The provider object with which to create this {@link FrequencyObservable}.
  */
-export const onEvery2Blocks$ = (() =>
-  onEveryBlock$().pipe(
+export const onEvery2Blocks$: FrequencyObservable<BigNumber> = (
+  provider?: any
+) =>
+  onEveryBlock$(provider).pipe(
     filter(n => +n % 2 === 0) // Around ~30s on mainnet // TODO Use isEqualTo and mod from bignumber.js
-  )) as FrequencyObservable<BigNumber>;
+  );
 onEvery2Blocks$.metadata = { name: 'onEvery2Blocks$' };
 
 /**
  * Observable that emits on every 4th block.
+ *
+ * @param provider - The provider object with which to create this {@link FrequencyObservable}.
  */
-export const onEvery4Blocks$ = (() =>
-  onEveryBlock$().pipe(
+export const onEvery4Blocks$: FrequencyObservable<BigNumber> = (
+  provider?: any
+) =>
+  onEveryBlock$(provider).pipe(
     filter(n => +n % 4 === 0) // Around ~1min on mainnet // TODO Use isEqualTo and mod from bignumber.js
-  )) as FrequencyObservable<BigNumber>;
+  );
 onEvery4Blocks$.metadata = { name: 'onEvery4Blocks$' };

packages/light.js/src/frequency/frequency.ts

@@ -0,0 +1,10 @@
+// Copyright 2015-2018 Parity Technologies (UK) Ltd.
+// This file is part of Parity.
+//
+// SPDX-License-Identifier: MIT
+
+export * from './accounts';
+export * from './blocks';
+export * from './health';
+export * from './other';
+export * from './time';

packages/light.js/src/frequency/health.ts

@@ -3,27 +3,22 @@
 //
 // SPDX-License-Identifier: MIT
 
-import api from '../api';
 import createPubsubObservable from './utils/createPubsubObservable';
 import { FrequencyObservable } from '../types';
 
 /**
  * Observable that emits when syncing status changes.
  */
 // TODO Pubsub doesn't exist on `net_peerCount`
-// export const onPeersChange$ = createPubsubObservable<number>('net_peerCount', api);
-// onPeersChange$.metadata = {
-//   calls: ['net_peerCount'],
-//   name: 'onPeersChange$'
-// };
 
 /**
  * Observable that emits when syncing status changes.
+ *
+ * @param provider - The provider object with which to create this {@link FrequencyObservable}.
  */
-export const onSyncingChanged$ = (() =>
-  createPubsubObservable('eth_syncing', api)) as FrequencyObservable<
-  object | boolean
->;
+export const onSyncingChanged$: FrequencyObservable<object | boolean> = (
+  provider?: any
+) => createPubsubObservable('eth_syncing', provider);
 onSyncingChanged$.metadata = {
   calls: ['eth_syncing'],
   name: 'onSyncingChanged$'

packages/light.js/src/frequency/index.ts

@@ -3,8 +3,7 @@
 //
 // SPDX-License-Identifier: MIT
 
-export * from './accounts';
-export * from './blocks';
-export * from './health';
-export * from './other';
-export * from './time';
+import * as frequency from './frequency';
+import { memoizeAll } from '../utils/memoizeAll';
+
+export default memoizeAll(frequency);

packages/light.js/src/frequency/other.ts

@@ -10,5 +10,5 @@ import { FrequencyObservable } from '../types';
 /**
  * Observable that emits only once.
  */
-export const onStartup$ = (() => of(0)) as FrequencyObservable<number>;
+export const onStartup$: FrequencyObservable<number> = (_?: any) => of(0);
 onStartup$.metadata = { name: 'onStartup$' };

packages/light.js/src/frequency/time.ts

@@ -10,23 +10,20 @@ import { FrequencyObservable } from '../types';
 /**
  * Observable that emits on every second.
  */
-export const onEverySecond$ = (() => timer(0, 1000)) as FrequencyObservable<
-  number
->;
+export const onEverySecond$: FrequencyObservable<number> = (_?: any) =>
+  timer(0, 1000);
 onEverySecond$.metadata = { name: 'onEverySecond$' };
 
 /**
  * Observable that emits on every other second.
  */
-export const onEvery2Seconds$ = (() => timer(0, 2000)) as FrequencyObservable<
-  number
->;
+export const onEvery2Seconds$: FrequencyObservable<number> = (_?: any) =>
+  timer(0, 2000);
 onEvery2Seconds$.metadata = { name: 'onEvery2Seconds$' };
 
 /**
  * Observable that emits every five seconds.
  */
-export const onEvery5Seconds$ = (() => timer(0, 5000)) as FrequencyObservable<
-  number
->;
+export const onEvery5Seconds$: FrequencyObservable<number> = (_?: any) =>
+  timer(0, 5000);
 onEvery5Seconds$.metadata = { name: 'onEvery5Seconds$' };

packages/light.js/src/frequency/utils/createPubsubObservable.spec.ts

@@ -8,37 +8,38 @@ import isObservable from '../../utils/isObservable';
 import { rejectApi, resolveApi } from '../../utils/testHelpers/mockApi';
 
 it('should return an Observable', () => {
-  expect(isObservable(createPubsubObservable('fake_method', resolveApi))).toBe(
-    true
-  );
+  expect(
+    isObservable(createPubsubObservable('fake_method', resolveApi()))
+  ).toBe(true);
 });
 
 it('should fire an event when pubsub publishes', done => {
-  createPubsubObservable('fake_method', resolveApi).subscribe(data => {
+  createPubsubObservable('fake_method', resolveApi()).subscribe(data => {
     expect(data).toBe('foo');
     done();
   });
 });
 
 it('should fire an error when pubsub errors', done => {
-  createPubsubObservable('fake_method', rejectApi).subscribe(null, err => {
+  createPubsubObservable('fake_method', rejectApi()).subscribe(null, err => {
     expect(err).toEqual(new Error('bar'));
     done();
   });
 });
 
-it('should fire an event when polling pubsub  publishes', done => {
-  createPubsubObservable('fake_method', () =>
-    resolveApi(undefined, false)
-  ).subscribe(data => {
-    expect(data).toBe('foo');
-    done();
-  });
+it('should fire an event when polling pubsub publishes', done => {
+  createPubsubObservable('fake_method', resolveApi('foo', false)).subscribe(
+    data => {
+      expect(data).toBe('foo');
+      done();
+    }
+  );
 });
 
 it('should fire an error when polling pubsub errors', done => {
-  createPubsubObservable('fake_method', () =>
-    rejectApi(undefined, false)
+  createPubsubObservable(
+    'fake_method',
+    rejectApi(new Error('bar'), false)
   ).subscribe(null, err => {
     expect(err).toEqual(new Error('bar'));
     done();

packages/light.js/src/frequency/utils/createPubsubObservable.ts

@@ -4,9 +4,11 @@
 // SPDX-License-Identifier: MIT
 
 import * as debug from 'debug';
+import { FrequencyObservable, FrequencyObservableMetadata } from '../../types';
 import { Observable, Observer, timer } from 'rxjs';
 import { switchMap } from 'rxjs/operators';
 
+import { createApiFromProvider, getApi } from '../../api';
 import { distinctReplayRefCount } from '../../utils/operators/distinctReplayRefCount';
 
 /**
@@ -17,27 +19,29 @@ import { distinctReplayRefCount } from '../../utils/operators/distinctReplayRefC
  */
 const createPubsubObservable = <T>(
   pubsub: string,
-  api: any // TODO @parity/api
-): Observable<T> => {
+  provider?: any // TODO @parity/api
+) => {
   const [namespace, method] = pubsub.split('_');
+  const api = provider ? createApiFromProvider(provider) : getApi();
 
   // There's a chance the provider doesn't support pubsub, for example
   // MetaMaskProvider. In this case, as suggested on their Github, the best
   // solution for now is to poll.
-  if (!api().isPubSub) {
+  if (!api.isPubSub) {
     debug('@parity/light.js:api')(
       `Pubsub not available for ${
-        api().provider
-          ? api().provider.constructor.name
-          : 'current Api provider'
-      }, polling "${pubsub}" every second.`
+        api.provider ? api.provider.constructor.name : 'current Api'
+      } provider, polling "${pubsub}" every second.`
     );
-    return timer(0, 1000).pipe(switchMap(() => api()[namespace][method]()));
+
+    return timer(0, 1000).pipe(
+      switchMap(() => api[namespace][method]())
+    ) as Observable<T>;
   }
 
   return Observable.create((observer: Observer<T>) => {
-    const subscription = api().pubsub[namespace][method](
-      (error: Error, result: T) => {
+    const subscription = api.pubsub[namespace][method](
+      (error: Error, result: any) => {
         // TODO use @parity/api type for result
         if (error) {
           observer.error(error);
@@ -48,9 +52,9 @@ const createPubsubObservable = <T>(
     );
     return () =>
       subscription.then((subscriptionId: string) =>
-        api().pubsub.unsubscribe(subscriptionId)
+        api.pubsub.unsubscribe(subscriptionId)
       );
-  }).pipe(distinctReplayRefCount());
+  }).pipe(distinctReplayRefCount()) as Observable<T>;
 };
 
 export default createPubsubObservable;

packages/light.js/src/rpc/eth/eth.ts

@@ -11,12 +11,7 @@ import { Address } from '../../types';
 import api from '../../api';
 import createRpc$ from '../utils/createRpc';
 import { isNullOrLoading, RPC_LOADING } from '../../utils/isLoading';
-import {
-  onAccountsChanged$,
-  onEveryBlock$,
-  onStartup$,
-  onSyncingChanged$
-} from '../../frequency';
+import frequency from '../../frequency';
 import { switchMapPromise } from '../../utils/operators';
 
 // /**
@@ -36,6 +31,7 @@ import { switchMapPromise } from '../../utils/operators';
 //     api
 //   );
 
+frequency.onAccountsChanged$;
 /**
  * Get the balance of a given account. Calls `eth_getBalance`.
  *
@@ -46,7 +42,7 @@ export const balanceOf$ = (address: Address, provider?: any) =>
   createRpc$<any, BigNumber>(
     {
       calls: ['eth_getBalance'],
-      frequency: [onEveryBlock$, onStartup$],
+      frequency: [frequency.onEveryBlock$, frequency.onStartup$],
       name: 'balanceOf$',
       pipes: api => [switchMapPromise(() => api.eth.getBalance(address))]
     },

packages/light.js/src/rpc/utils/createRpc.ts

@@ -52,7 +52,7 @@ const frequencyMixins = {
  */
 const createRpc = <Source, Out>(
   metadata: Metadata<Source, Out>,
-  provider: any
+  provider?: any
 ) => {
   const api = provider ? new Api(provider) : getApi();
   // rpc$ will hold the RpcObservable minus its metadata
@@ -62,7 +62,7 @@ const createRpc = <Source, Out>(
     // FrequencyObservables
     const source$ = metadata.dependsOn
       ? metadata.dependsOn(...args)
-      : merge(...metadata.frequency.map(f => f()));
+      : merge(...metadata.frequency.map(f => f(provider)));
 
     // The last arguments is an options, if it's an object
     // TODO What if we pass a single object as argument, which is not options?