Don't use Parity-specific JSONRPCs

packages/api/src/transport/ws/ws.js

@@ -212,11 +212,11 @@ class Ws extends JsonRpcBase {
   }
 
   _extract (result) {
-    const { result: res, id, method, params } = result;
+    const { result: res, error, id, method, params } = result;
     const msg = this._messages[id];
 
     // initial pubsub ACK
-    if (id && msg.subscription) {
+    if (id && msg.subscription && !error) {
       // save subscription to map subId -> messageId
       this._subscriptions[msg.subscription] = this._subscriptions[msg.subscription] || {};
       this._subscriptions[msg.subscription][res] = id;

packages/light.js/example/src/provider.js

@@ -7,6 +7,9 @@ import Api from '@parity/api';
 
 export const currentProvider = window.web3 && window.web3.currentProvider;
 export const localProvider = new Api.Provider.Ws('ws://127.0.0.1:8546');
+export const infuraProvider = new Api.Provider.Ws(
+  'wss://mainnet.infura.io/_ws'
+);
 
 const provider = currentProvider || localProvider;
 

packages/light.js/src/api.ts

@@ -25,11 +25,6 @@ export const createApiFromProvider = memoizee(
  */
 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.`
-    );
-  }
 };
 
 /**

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

@@ -12,7 +12,11 @@ import createPubsubObservable from './utils/createPubsubObservable';
  * @param options - Options to pass to {@link FrequencyObservable}.
  */
 export function onAccountsChanged$ (options?: FrequencyObservableOptions) {
-  return createPubsubObservable<Address[]>('eth_accounts', options);
+  return createPubsubObservable<Address[]>(
+    'eth_accounts',
+    'eth_accounts',
+    options
+  );
 }
 
 /**
@@ -21,5 +25,9 @@ export function onAccountsChanged$ (options?: FrequencyObservableOptions) {
  * @param options - Options to pass to {@link FrequencyObservable}.
  */
 export function onAccountsInfoChanged$ (options?: FrequencyObservableOptions) {
-  return createPubsubObservable<AccountsInfo>('parity_accountsInfo', options);
+  return createPubsubObservable<AccountsInfo>(
+    'parity_accountsInfo',
+    'parity_accountsInfo',
+    options
+  );
 }

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

@@ -3,14 +3,14 @@
 //
 // SPDX-License-Identifier: MIT
 
-import BigNumber from 'bignumber.js';
-import { filter, map, withLatestFrom } from 'rxjs/operators';
+import { filter, map, startWith, withLatestFrom } from 'rxjs/operators';
 import * as memoizee from 'memoizee';
 import { Observable } from 'rxjs';
 
+import { Block, FrequencyObservableOptions } from '../types';
 import { createApiFromProvider, getApi } from '../api';
 import createPubsubObservable from './utils/createPubsubObservable';
-import { FrequencyObservableOptions } from '../types';
+import { distinctValues } from '../utils/operators';
 import { onSyncingChanged$ } from './health';
 
 /**
@@ -20,11 +20,16 @@ import { onSyncingChanged$ } from './health';
  */
 const onEveryBlockWithApi$ = memoizee(
   (api: any, options?: FrequencyObservableOptions) =>
-    createPubsubObservable('eth_blockNumber', options).pipe(
-      withLatestFrom(onSyncingChanged$(options)),
+    createPubsubObservable(
+      'eth_newHeads',
+      'eth_getBlockByNumber',
+      options
+    ).pipe(
+      withLatestFrom(onSyncingChanged$(options).pipe(startWith(false))),
       filter(([_, isSyncing]) => isSyncing === false),
-      map(([blockNumber]) => blockNumber)
-    ) as Observable<BigNumber>,
+      map(([blockNumber]) => blockNumber),
+      distinctValues()
+    ) as Observable<Block>,
   { length: 1 } // Only memoize by api
 );
 

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

@@ -17,5 +17,9 @@ import { FrequencyObservableOptions } from '../types';
  * @param options - Options to pass to {@link FrequencyObservable}.
  */
 export function onSyncingChanged$ (options?: FrequencyObservableOptions) {
-  return createPubsubObservable<object | false>('eth_syncing', options);
+  return createPubsubObservable<object | false>(
+    'eth_syncing',
+    'eth_syncing',
+    options
+  );
 }

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

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

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

@@ -4,22 +4,45 @@
 // SPDX-License-Identifier: MIT
 
 import * as debug from 'debug';
+import { exhaustMap } from 'rxjs/operators';
 import { FrequencyObservableOptions } from '../../types';
 import * as memoizee from 'memoizee';
 import { Observable, Observer, timer } from 'rxjs';
-import { switchMap } from 'rxjs/operators';
 
 import { createApiFromProvider, getApi } from '../../api';
 import { distinctReplayRefCount } from '../../utils/operators/distinctReplayRefCount';
 
+const POLL_INTERVAL = 1000;
+
+/**
+ * Create a polling function, calls the `fallback` JSONRPC on each second, or
+ * on previous call's result, whichever comes last.
+ *
+ * @ignore
+ */
+function createPoll<T> (
+  fallback: string,
+  api: any,
+  pollInterval = POLL_INTERVAL
+) {
+  const [fallbackNamespace, fallbackMethod] = fallback.split('_');
+
+  return timer(0, pollInterval).pipe(
+    exhaustMap(() => api[fallbackNamespace][fallbackMethod]())
+  ) as Observable<T>;
+}
+
 /**
  * Given an api, returns an Observable that emits on each pubsub event.
  * Pure function version of {@link createPubsubObservable}.
  *
  * @ignore
+ * @param pubsub - The pubsub method to subscribe to.
+ * @param fallback - If pubsub doesn't work, poll this method every
+ * POLL_INTERVAL ms.
  */
 const createPubsubObservableWithApi = memoizee(
-  <T>(pubsub: string, api: any) => {
+  <T>(pubsub: string, fallback: string, api: any) => {
     const [namespace, method] = pubsub.split('_');
 
     // There's a chance the provider doesn't support pubsub, for example
@@ -29,12 +52,10 @@ const createPubsubObservableWithApi = memoizee(
       debug('@parity/light.js:api')(
         `Pubsub not available for ${
           api.provider ? api.provider.constructor.name : 'current Api'
-        } provider, polling "${pubsub}" every second.`
+        } provider, polling "${fallback}" every ${POLL_INTERVAL}ms.`
       );
 
-      return timer(0, 1000).pipe(
-        switchMap(() => api[namespace][method]())
-      ) as Observable<T>;
+      return createPoll<T>(fallback, api);
     }
 
     return Observable.create((observer: Observer<T>) => {
@@ -47,10 +68,25 @@ const createPubsubObservableWithApi = memoizee(
             observer.next(result);
           }
         }
-      );
+      ).catch(() => {
+        // If we get an error during subscription, then default to fallback.
+        // TODO Should this be done on @parity/api?
+        debug('@parity/light.js:api')(
+          `Pubsub not available for method "${pubsub}", polling "${fallback}" every ${POLL_INTERVAL}ms`
+        );
+
+        createPoll<T>(fallback, api).subscribe(
+          e => observer.next(e),
+          e => observer.error(e),
+          () => observer.complete()
+        );
+      });
+
       return () =>
         subscription.then((subscriptionId: string) =>
-          api.pubsub.unsubscribe(subscriptionId)
+          subscriptionId
+            ? api.pubsub.unsubscribe(subscriptionId)
+            : Promise.resolve()
         );
     }).pipe(distinctReplayRefCount()) as Observable<T>;
   }
@@ -64,11 +100,12 @@ const createPubsubObservableWithApi = memoizee(
  */
 const createPubsubObservable = <T>(
   pubsub: string,
+  fallback: string,
   { provider }: FrequencyObservableOptions = {}
 ) => {
   const api = provider ? createApiFromProvider(provider) : getApi();
 
-  return createPubsubObservableWithApi<T>(pubsub, api);
+  return createPubsubObservableWithApi<T>(pubsub, fallback, api);
 };
 
 export default createPubsubObservable;

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

@@ -7,7 +7,7 @@ import BigNumber from 'bignumber.js';
 import { of } from 'rxjs';
 import { map, switchMap } from 'rxjs/operators';
 
-import { Address, RpcObservableOptions } from '../types';
+import { Address, Block, RpcObservableOptions } from '../types';
 import createRpc$ from './utils/createRpc';
 import frequency from '../frequency';
 import { isNullOrLoading, RPC_LOADING } from '../utils/isLoading';
@@ -66,9 +66,10 @@ export function defaultAccount$ (options?: RpcObservableOptions) {
  * @return {Observable<Number>} - An Observable containing the block height.
  */
 export function blockNumber$ (options?: RpcObservableOptions) {
-  return createRpc$<BigNumber, BigNumber>({
+  return createRpc$<Block, BigNumber>({
     frequency: [frequency.onEveryBlock$],
-    name: 'blockNumber$'
+    name: 'blockNumber$',
+    pipes: () => [map(block => block.number)]
   })(options)();
 }
 
@@ -81,11 +82,10 @@ export function myBalance$ (options?: RpcObservableOptions) {
     dependsOn: defaultAccount$,
     name: 'myBalance$',
     pipes: () => [
-      switchMap(
-        defaultAccount =>
-          isNullOrLoading(defaultAccount)
-            ? of(RPC_LOADING)
-            : balanceOf$(defaultAccount)
+      switchMap(defaultAccount =>
+        isNullOrLoading(defaultAccount)
+          ? of(RPC_LOADING)
+          : balanceOf$(defaultAccount)
       )
     ]
   })(options)();

packages/light.js/src/types.ts

@@ -26,6 +26,11 @@ export type Address = string;
 // TODO This should be on @parity/api
 export type ApiValue = any;
 
+// TODO This should be on @parity/api
+export type Block = {
+  number: BigNumber;
+};
+
 export interface Metadata<Source, Out> {
   calledWithArgs?: {
     [key: string]: ReplaySubject<Out>;

packages/light.js/src/utils/operators/distinctValues.spec.ts

@@ -52,3 +52,4 @@ testValue(2, 'number');
 testValue('foo', 'string');
 testValue({ foo: 'bar' }, 'object');
 testValue(new BigNumber(2), 'BigNumber');
+testValue({ number: new BigNumber(2) }, 'Block');

packages/light.js/src/utils/operators/distinctValues.ts

@@ -7,18 +7,37 @@ import BigNumber from 'bignumber.js';
 import { distinctUntilChanged } from 'rxjs/operators';
 import { isObject } from '@parity/api/lib/util/types';
 
+import { Block } from '../../types';
+
 /**
  * An intelligent distinctUntilChanged().
  *
  * @ignore
  */
 export const distinctValues = <T>() =>
   distinctUntilChanged<T>((x, y) => {
+    // If T == Block
+    if (
+      x &&
+      y &&
+      ((x as unknown) as Block).number &&
+      ((y as unknown) as Block).number
+    ) {
+      return ((x as unknown) as Block).number.eq(
+        ((y as unknown) as Block).number
+      );
+    }
+
+    // If T == BigNumber
     if (BigNumber.isBigNumber(x) && BigNumber.isBigNumber(y)) {
-      return ((x as any) as BigNumber).eq((y as any) as BigNumber);
+      return ((x as unknown) as BigNumber).eq((y as unknown) as BigNumber);
     }
+
+    // If T == object
     if (isObject(x) && isObject(y)) {
       return JSON.stringify(x) === JSON.stringify(y); // TODO Do a deep equal instead
     }
+
+    // Other cases
     return x === y;
   });

packages/light.js/src/utils/testHelpers/mockApi.ts

@@ -22,7 +22,7 @@ export class MockProvider extends EventEmitter {
 
 // List of JSONRPCs we want to mock
 const listOfMockRps: { [index: string]: string[] } = {
-  eth: ['accounts', 'blockNumber', 'getBalance', 'syncing'],
+  eth: ['accounts', 'blockNumber', 'getBalance', 'newHeads', 'syncing'],
   fake: ['method'],
   net: ['peerCount'],
   parity: ['accountsInfo', 'chain', 'postTransaction']