From d45a9cfe587dae693dbac987a741cbb97e40d2bc Mon Sep 17 00:00:00 2001 From: Jaco Date: Tue, 5 Oct 2021 08:57:30 +0300 Subject: [PATCH] Rework derives to ensure cached registries (#4021) --- packages/api-derive/src/chain/getBlock.ts | 24 ++++++++++------- packages/api-derive/src/chain/getHeader.ts | 25 +++++++++++------- .../src/chain/subscribeNewBlocks.ts | 26 +++++++++---------- packages/api-derive/src/tx/events.ts | 22 +++++++++------- 4 files changed, 55 insertions(+), 42 deletions(-) diff --git a/packages/api-derive/src/chain/getBlock.ts b/packages/api-derive/src/chain/getBlock.ts index 11345cee7b..9872f6b08f 100644 --- a/packages/api-derive/src/chain/getBlock.ts +++ b/packages/api-derive/src/chain/getBlock.ts @@ -5,7 +5,7 @@ import type { Observable } from 'rxjs'; import type { ApiInterfaceRx } from '@polkadot/api/types'; import type { SignedBlockExtended } from '../type/types'; -import { catchError, combineLatest, map, of } from 'rxjs'; +import { catchError, combineLatest, map, of, switchMap } from 'rxjs'; import { createSignedBlockExtended } from '../type'; import { memo } from '../util'; @@ -25,15 +25,19 @@ import { memo } from '../util'; */ export function getBlock (instanceId: string, api: ApiInterfaceRx): (hash: Uint8Array | string) => Observable { return memo(instanceId, (hash: Uint8Array | string): Observable => - combineLatest([ - api.rpc.chain.getBlock(hash), - api.query.system.events.at(hash), - api.query.session - ? api.query.session.validators.at(hash) - : of([]) - ]).pipe( - map(([signedBlock, events, validators]): SignedBlockExtended => - createSignedBlockExtended(api.registry, signedBlock, events, validators) + // we get the block first, setting up the registry + api.rpc.chain.getBlock(hash).pipe( + switchMap((signedBlock) => + combineLatest([ + api.query.system.events.at(hash), + api.query.session + ? api.query.session.validators.at(hash) + : of([]) + ]).pipe( + map(([events, validators]) => + createSignedBlockExtended(api.registry, signedBlock, events, validators) + ) + ) ), catchError((): Observable => // where rpc.chain.getHeader throws, we will land here - it can happen that diff --git a/packages/api-derive/src/chain/getHeader.ts b/packages/api-derive/src/chain/getHeader.ts index 6a0f16ae64..9bd705e6e0 100644 --- a/packages/api-derive/src/chain/getHeader.ts +++ b/packages/api-derive/src/chain/getHeader.ts @@ -3,13 +3,20 @@ import type { Observable } from 'rxjs'; import type { ApiInterfaceRx } from '@polkadot/api/types'; +import type { AccountId } from '@polkadot/types/interfaces'; import type { HeaderExtended } from '../type/types'; -import { catchError, combineLatest, map, of } from 'rxjs'; +import { catchError, map, of, switchMap } from 'rxjs'; import { createHeaderExtended } from '../type'; import { memo } from '../util'; +function _getValidators (api: ApiInterfaceRx, hash: Uint8Array | string): Observable { + return api.query.session + ? api.query.session.validators.at(hash) + : of([] as AccountId[]); +} + /** * @name getHeader * @param {( Uint8Array | string )} hash - A block hash as U8 array or string. @@ -26,14 +33,14 @@ import { memo } from '../util'; */ export function getHeader (instanceId: string, api: ApiInterfaceRx): (hash: Uint8Array | string) => Observable { return memo(instanceId, (hash: Uint8Array | string): Observable => - combineLatest([ - api.rpc.chain.getHeader(hash), - api.query.session - ? api.query.session.validators.at(hash) - : of([]) - ]).pipe( - map(([header, validators]): HeaderExtended => - createHeaderExtended(header.registry, header, validators) + // we get the header first, setting up the registry + api.rpc.chain.getHeader(hash).pipe( + switchMap((header) => + _getValidators(api, hash).pipe( + map((validators) => + createHeaderExtended(header.registry, header, validators) + ) + ) ), catchError((): Observable => // where rpc.chain.getHeader throws, we will land here - it can happen that diff --git a/packages/api-derive/src/chain/subscribeNewBlocks.ts b/packages/api-derive/src/chain/subscribeNewBlocks.ts index f7002d4003..6d63251581 100644 --- a/packages/api-derive/src/chain/subscribeNewBlocks.ts +++ b/packages/api-derive/src/chain/subscribeNewBlocks.ts @@ -3,11 +3,9 @@ import type { Observable } from 'rxjs'; import type { ApiInterfaceRx } from '@polkadot/api/types'; -import type { Vec } from '@polkadot/types'; -import type { EventRecord, SignedBlock } from '@polkadot/types/interfaces'; -import type { HeaderExtended, SignedBlockExtended } from '../type/types'; +import type { SignedBlockExtended } from '../type/types'; -import { combineLatest, map, of, switchMap } from 'rxjs'; +import { map, switchMap } from 'rxjs'; import { createSignedBlockExtended } from '../type'; import { memo } from '../util'; @@ -19,18 +17,20 @@ import { memo } from '../util'; export function subscribeNewBlocks (instanceId: string, api: ApiInterfaceRx): () => Observable { return memo(instanceId, (): Observable => api.derive.chain.subscribeNewHeads().pipe( - switchMap((header): Observable<[SignedBlock, Vec, HeaderExtended]> => { + switchMap((header) => { const blockHash = header.createdAtHash || header.hash; - return combineLatest( - api.rpc.chain.getBlock(blockHash), - api.query.system.events.at(blockHash), - of(header) + // we get the block first, setting up the registry + return api.rpc.chain.getBlock(blockHash).pipe( + switchMap((block) => + api.query.system.events.at(blockHash).pipe( + map((events) => + createSignedBlockExtended(block.registry, block, events, header.validators) + ) + ) + ) ); - }), - map(([block, events, header]) => - createSignedBlockExtended(block.registry, block, events, header.validators) - ) + }) ) ); } diff --git a/packages/api-derive/src/tx/events.ts b/packages/api-derive/src/tx/events.ts index 7a468ff433..d64df5b426 100644 --- a/packages/api-derive/src/tx/events.ts +++ b/packages/api-derive/src/tx/events.ts @@ -5,7 +5,7 @@ import type { Observable } from 'rxjs'; import type { ApiInterfaceRx } from '@polkadot/api/types'; import type { EventRecord, Hash, SignedBlock } from '@polkadot/types/interfaces'; -import { combineLatest, map } from 'rxjs'; +import { map, switchMap } from 'rxjs'; import { memo } from '../util'; @@ -15,15 +15,17 @@ interface Result { } export function events (instanceId: string, api: ApiInterfaceRx): (at: Hash) => Observable { - return memo(instanceId, (at: Hash): Observable => - combineLatest([ - api.query.system.events.at(at), - api.rpc.chain.getBlock(at) - ]).pipe( - map(([events, block]) => ({ - block, - events - })) + return memo(instanceId, (at: Hash) => + // we get the block first, setting up the registry + api.rpc.chain.getBlock(at).pipe( + switchMap((block) => + api.query.system.events.at(at).pipe( + map((events): Result => ({ + block, + events + })) + ) + ) ) ); }