Add derive.chain.{getHeader,subscribeNewHead} with author (#602)

* Add derive.chain.{getHeader,subscribeNewHead} with author

* tslint cast, ignore

packages/api-derive/src/chain/getHeader.ts

@@ -0,0 +1,35 @@
+// 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 { Observable, combineLatest, of } from 'rxjs';
+import { catchError, map } from 'rxjs/operators';
+import { ApiInterface$Rx } from '@polkadot/api/types';
+import { AccountId, Header } from '@polkadot/types/index';
+import { HeaderExtended } from '@polkadot/types/Header';
+
+import { drr } from '../util/drr';
+import { HeaderAndValidators } from './subscribeNewHead';
+
+/**
+ * Get the a specific block header and extend it with the author
+ */
+export function getHeader (api: ApiInterface$Rx) {
+  return (hash: Uint8Array | string): Observable<HeaderExtended | undefined> =>
+    combineLatest(
+      api.rpc.chain.getHeader(hash) as Observable<Header>,
+      api.query.session.validators.at(hash) as any as Observable<Array<AccountId>>
+    ).pipe(
+      map(([header, validators]: HeaderAndValidators) =>
+        new HeaderExtended(header, validators)
+      ),
+      catchError(() =>
+        // where rpc.chain.getHeader throws, we will land here - it can happen that
+        // we supplied an invalid hash. (Due to defaults, storeage will have an
+        // empty value, so only the RPC is affected). So return undefined
+        // tslint:disable-next-line
+        of() as Observable<undefined>
+      ),
+      drr()
+    );
+}

packages/api-derive/src/chain/index.ts

@@ -4,3 +4,5 @@
 
 export * from './bestNumber';
 export * from './bestNumberFinalised';
+export * from './getHeader';
+export * from './subscribeNewHead';

packages/api-derive/src/chain/subscribeNewHead.ts

@@ -0,0 +1,39 @@
+// 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 { Observable, combineLatest, of } from 'rxjs';
+import { filter, map, switchMap } from 'rxjs/operators';
+import { ApiInterface$Rx } from '@polkadot/api/types';
+import { AccountId, Header } from '@polkadot/types/index';
+import { HeaderExtended } from '@polkadot/types/Header';
+
+import { drr } from '../util/drr';
+
+export type HeaderAndValidators = [Header, Array<AccountId>];
+
+/**
+ * Subscribe to block headers and extend it with the author
+ */
+export function subscribeNewHead (api: ApiInterface$Rx) {
+  return (): Observable<HeaderExtended> =>
+    (api.rpc.chain.subscribeNewHead() as Observable<Header>)
+      .pipe(
+        filter((header: Header) =>
+          header && !!header.blockNumber
+        ),
+        switchMap((header: Header): Observable<HeaderAndValidators> =>
+          combineLatest(
+            of(header),
+            // theoretically we could combine at the first call with session.validators(), however
+            // we make 100% sure we actually get the validators at a specific block so when these
+            // change at an era boundary, we have the previous values to ensure our indexes are correct
+            api.query.session.validators.at(header.hash) as any as Observable<Array<AccountId>>
+          )
+        ),
+        map(([header, validators]: HeaderAndValidators) =>
+          new HeaderExtended(header, validators)
+        ),
+        drr()
+      );
+}