Merge pull request #36 from paritytech/am-every-block

Make onEveryBlock$ dormant

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

@@ -4,15 +4,41 @@
 // SPDX-License-Identifier: MIT
 
 import BigNumber from 'bignumber.js';
+import { filter, map, withLatestFrom } from 'rxjs/operators';
+import * as memoizee from 'memoizee';
+import { Observable } from 'rxjs';
 
+import { createApiFromProvider, getApi } from '../api';
 import createPubsubObservable from './utils/createPubsubObservable';
 import { FrequencyObservableOptions } from '../types';
+import { onSyncingChanged$ } from './health';
 
 /**
- * Observable that emits on every new block.
+ * Pure function version of {@link onEveryBlock$}.
+ *
+ * @ignore
+ */
+const onEveryBlockWithApi$ = memoizee(
+  (api: any, options: FrequencyObservableOptions) =>
+    onSyncingChanged$(options).pipe(
+      filter((isSyncing: boolean) => isSyncing === false),
+      withLatestFrom(createPubsubObservable('eth_blockNumber', options)),
+      map(([_, blockNumber]) => blockNumber)
+    ) as Observable<BigNumber>,
+  { length: 1 } // Only memoize by api
+);
+
+/**
+ * Observable that emits on every new block. Note: this FrequencyObservable
+ * won't fire when the light client is syncing.
  *
  * @param options - Options to pass to {@link FrequencyObservable}.
  */
 export function onEveryBlock$(options?: FrequencyObservableOptions) {
-  return createPubsubObservable<BigNumber>('eth_blockNumber', options);
+  const api =
+    options && options.provider
+      ? createApiFromProvider(options.provider)
+      : getApi();
+
+  return onEveryBlockWithApi$(api, options);
 }

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

@@ -43,7 +43,7 @@ const testRpc = (name: string, rpc$: RpcObservable<any, any>) =>
 
     it('result Observable should return values', done => {
       rpc$({}).subscribe(data => {
-        expect(data).toBeTruthy();
+        expect(data).not.toBeNull();
         done();
       });
     });

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

@@ -65,6 +65,12 @@ const createApi = (
             };
       });
 
+      // For eth.syncing pubsub, always return false
+      apiObject.pubsub.eth.syncing = (callback: Function) => {
+        callback(null, false);
+        return Promise.resolve(1);
+      };
+
       return apiObject;
     },
     {