Use asyncScheduler for all derives (#2043)

packages/api-derive/src/staking/account.ts

@@ -7,9 +7,8 @@ import { Balance, BlockNumber, StakingLedger, UnlockChunk } from '@polkadot/type
 import { DerivedSessionInfo, DerivedStakingAccount, DerivedStakingQuery, DerivedUnlocking } from '../types';
 
 import BN from 'bn.js';
-import { combineLatest, Observable } from 'rxjs';
-import { map } from 'rxjs/operators';
-import { createType } from '@polkadot/types';
+import { Observable, asyncScheduler, combineLatest } from 'rxjs';
+import { map, observeOn } from 'rxjs/operators';
 
 import { isUndefined } from '@polkadot/util';
 
@@ -33,7 +32,7 @@ function remainingBlocks (api: ApiInterfaceRx, era: BN, sessionInfo: DerivedSess
   const remaining = era.sub(sessionInfo.currentEra);
 
   // on the Rust side the current-era >= era-for-unlock (removal done on >)
-  return createType(api.registry, 'BlockNumber', remaining.gtn(0)
+  return api.registry.createType('BlockNumber', remaining.gtn(0)
     ? remaining
       .subn(1)
       .mul(sessionInfo.eraLength)
@@ -58,7 +57,7 @@ function calculateUnlocking (api: ApiInterfaceRx, stakingLedger: StakingLedger |
   // group the unlock chunks that have the same era and sum their values
   const groupedResult = groupByEra(unlockingChunks);
   const results = Object.entries(groupedResult).map(([eraString, value]): DerivedUnlocking => ({
-    value: createType(api.registry, 'Balance', value),
+    value: api.registry.createType('Balance', value),
     remainingBlocks: remainingBlocks(api, new BN(eraString), sessionInfo)
   }));
 
@@ -67,10 +66,10 @@ function calculateUnlocking (api: ApiInterfaceRx, stakingLedger: StakingLedger |
 
 function redeemableSum (api: ApiInterfaceRx, stakingLedger: StakingLedger | undefined, sessionInfo: DerivedSessionInfo): Balance {
   if (isUndefined(stakingLedger)) {
-    return createType(api.registry, 'Balance');
+    return api.registry.createType('Balance');
   }
 
-  return createType(api.registry, 'Balance', stakingLedger.unlocking.reduce((total, { era, value }): BN => {
+  return api.registry.createType('Balance', stakingLedger.unlocking.reduce((total, { era, value }): BN => {
     return remainingBlocks(api, era.unwrap(), sessionInfo).eqn(0)
       ? total.add(value.unwrap())
       : total;
@@ -94,6 +93,7 @@ export function account (api: ApiInterfaceRx): (accountId: Uint8Array | string)
       api.derive.session.info(),
       api.derive.staking.query(accountId)
     ]).pipe(
+      observeOn(asyncScheduler),
       map(([sessionInfo, query]: [DerivedSessionInfo, DerivedStakingQuery]) =>
         parseResult(api, sessionInfo, query))
     ));

packages/api-derive/src/staking/controllers.ts

@@ -6,8 +6,8 @@ import { ApiInterfaceRx } from '@polkadot/api/types';
 import { AccountId, ValidatorPrefs } from '@polkadot/types/interfaces';
 import { ITuple } from '@polkadot/types/types';
 
-import { Observable, combineLatest, of } from 'rxjs';
-import { switchMap } from 'rxjs/operators';
+import { Observable, asyncScheduler, combineLatest, of } from 'rxjs';
+import { observeOn, switchMap } from 'rxjs/operators';
 import { Option, Vec } from '@polkadot/types';
 
 import { memo } from '../util';
@@ -20,6 +20,7 @@ type DeriveControllers = [AccountId[], Option<AccountId>[]];
 export function controllers (api: ApiInterfaceRx): () => Observable<DeriveControllers> {
   return memo((): Observable<[AccountId[], Option<AccountId>[]]> =>
     api.query.staking.validators<ITuple<[Vec<AccountId>, Vec<ValidatorPrefs>]>>().pipe(
+      observeOn(asyncScheduler),
       switchMap(([stashIds]): Observable<DeriveControllers> =>
         combineLatest([
           of(stashIds),
@@ -28,6 +29,7 @@ export function controllers (api: ApiInterfaceRx): () => Observable<DeriveContro
             ? of([])
             : api.query.staking.bonded.multi<Option<AccountId>>(stashIds)
         ])
-      )
+      ),
+      observeOn(asyncScheduler)
     ));
 }

packages/api-derive/src/staking/electedInfo.ts

@@ -6,22 +6,26 @@ import { ApiInterfaceRx } from '@polkadot/api/types';
 import { AccountId } from '@polkadot/types/interfaces';
 import { DerivedStakingQuery, DerivedStakingElected } from '../types';
 
-import { Observable, combineLatest, of } from 'rxjs';
-import { map, switchMap } from 'rxjs/operators';
+import { Observable, asyncScheduler, combineLatest, of } from 'rxjs';
+import { map, observeOn, switchMap } from 'rxjs/operators';
 
 import { memo } from '../util';
 
 export function electedInfo (api: ApiInterfaceRx): () => Observable<DerivedStakingElected> {
   return memo((): Observable<DerivedStakingElected> =>
     api.derive.staking.validators().pipe(
+      observeOn(asyncScheduler),
       switchMap(({ nextElected }): Observable<[AccountId[], DerivedStakingQuery[]]> =>
         combineLatest([
           of(nextElected),
-          combineLatest(nextElected.map((accountId): Observable<DerivedStakingQuery> =>
-            api.derive.staking.query(accountId)
-          ))
+          combineLatest(
+            nextElected.map((accountId): Observable<DerivedStakingQuery> =>
+              api.derive.staking.query(accountId)
+            )
+          )
         ])
       ),
+      observeOn(asyncScheduler),
       map(([nextElected, info]): DerivedStakingElected => ({
         info,
         nextElected

packages/api-derive/src/staking/erasExposure.ts

@@ -7,8 +7,8 @@ import { EraIndex, Exposure } from '@polkadot/types/interfaces';
 import { DeriveEraExposure, DeriveEraNominatorExposure, DeriveEraValidatorExposure } from '../types';
 
 import BN from 'bn.js';
-import { Observable, combineLatest, of } from 'rxjs';
-import { map, switchMap, take } from 'rxjs/operators';
+import { Observable, asyncScheduler, combineLatest, of } from 'rxjs';
+import { map, observeOn, switchMap, take } from 'rxjs/operators';
 import { StorageKey } from '@polkadot/types';
 
 import { memo } from '../util';
@@ -38,6 +38,7 @@ function mapStakers (era: EraIndex, stakers: KeysAndExposures): DeriveEraExposur
 export function eraExposure (api: ApiInterfaceRx): (era: EraIndex) => Observable<DeriveEraExposure> {
   return memo((era: EraIndex): Observable<DeriveEraExposure> =>
     api.query.staking.erasStakersClipped.entries(era).pipe(
+      observeOn(asyncScheduler),
       take(1),
       map((stakers) => mapStakers(era, stakers))
     )
@@ -47,9 +48,12 @@ export function eraExposure (api: ApiInterfaceRx): (era: EraIndex) => Observable
 export function erasExposure (api: ApiInterfaceRx): (withActive?: boolean | BN | number) => Observable<DeriveEraExposure[]> {
   return memo((withActive?: boolean | BN | number): Observable<DeriveEraExposure[]> =>
     api.derive.staking.erasHistoric(withActive).pipe(
+      observeOn(asyncScheduler),
       switchMap((eras): Observable<DeriveEraExposure[]> =>
         eras.length
-          ? combineLatest(eras.map((era) => api.derive.staking.eraExposure(era)))
+          ? combineLatest(
+            eras.map((era) => api.derive.staking.eraExposure(era))
+          )
           : of([])
       )
     )

packages/api-derive/src/staking/erasPoints.ts

@@ -7,8 +7,8 @@ import { EraIndex, EraRewardPoints } from '@polkadot/types/interfaces';
 import { DeriveEraPoints, DeriveEraValPoints } from '../types';
 
 import BN from 'bn.js';
-import { Observable, combineLatest, of } from 'rxjs';
-import { map, switchMap } from 'rxjs/operators';
+import { Observable, asyncScheduler, combineLatest, of } from 'rxjs';
+import { map, observeOn, switchMap } from 'rxjs/operators';
 
 import { memo } from '../util';
 
@@ -25,6 +25,7 @@ function mapValidators ({ individual }: EraRewardPoints): DeriveEraValPoints {
 export function erasPoints (api: ApiInterfaceRx): (withActive?: boolean | BN | number) => Observable<DeriveEraPoints[]> {
   return memo((withActive?: boolean | number): Observable<DeriveEraPoints[]> =>
     api.derive.staking.erasHistoric(withActive).pipe(
+      observeOn(asyncScheduler),
       switchMap((eras): Observable<[EraIndex[], EraRewardPoints[]]> =>
         combineLatest([
           of(eras),
@@ -33,6 +34,7 @@ export function erasPoints (api: ApiInterfaceRx): (withActive?: boolean | BN | n
             : of([])
         ])
       ),
+      observeOn(asyncScheduler),
       map(([eras, points]): DeriveEraPoints[] =>
         eras.map((era, index): DeriveEraPoints => ({
           era,

packages/api-derive/src/staking/erasRewards.ts

@@ -7,8 +7,8 @@ import { Balance, EraIndex } from '@polkadot/types/interfaces';
 import { DeriveEraRewards } from '../types';
 
 import BN from 'bn.js';
-import { Observable, combineLatest, of } from 'rxjs';
-import { map, switchMap } from 'rxjs/operators';
+import { Observable, asyncScheduler, combineLatest, of } from 'rxjs';
+import { map, observeOn, switchMap } from 'rxjs/operators';
 import { Option } from '@polkadot/types';
 
 import { memo } from '../util';
@@ -16,6 +16,7 @@ import { memo } from '../util';
 export function erasRewards (api: ApiInterfaceRx): (withActive?: boolean | BN | number) => Observable<DeriveEraRewards[]> {
   return memo((withActive?: boolean | BN | number): Observable<DeriveEraRewards[]> =>
     api.derive.staking.erasHistoric(withActive).pipe(
+      observeOn(asyncScheduler),
       switchMap((eras): Observable<[EraIndex[], Option<Balance>[]]> =>
         combineLatest([
           of(eras),
@@ -24,6 +25,7 @@ export function erasRewards (api: ApiInterfaceRx): (withActive?: boolean | BN |
             : of([])
         ])
       ),
+      observeOn(asyncScheduler),
       map(([eras, rewards]): DeriveEraRewards[] =>
         eras.map((era, index): DeriveEraRewards => ({
           era,

packages/api-derive/src/staking/overview.ts

@@ -6,20 +6,19 @@ import { AccountId, EraIndex, EraPoints, EraRewardPoints, RewardPoint } from '@p
 import { ApiInterfaceRx } from '@polkadot/api/types';
 import { DerivedStakingOverview } from '../types';
 
-import { Observable, combineLatest, of } from 'rxjs';
-import { map, switchMap } from 'rxjs/operators';
-import { createType } from '@polkadot/types';
+import { Observable, asyncScheduler, combineLatest, of } from 'rxjs';
+import { map, observeOn, switchMap } from 'rxjs/operators';
 
 import { memo } from '../util';
 
 function retrievePointsPrev (api: ApiInterfaceRx, activeEra: EraIndex, currentElected: AccountId[]): Observable<EraRewardPoints> {
   return api.query.staking.currentEraPointsEarned<EraPoints>(activeEra).pipe(
     map(({ individual, total }): EraRewardPoints =>
-      createType(api.registry, 'EraRewardPoints', {
+      api.registry.createType('EraRewardPoints', {
         total,
         individual: new Map<AccountId, RewardPoint>(
           individual
-            .map((points): RewardPoint => createType(api.registry, 'RewardPoint', points))
+            .map((points): RewardPoint => api.registry.createType('RewardPoint', points))
             .map((points, index): [AccountId, RewardPoint] => [currentElected[index], points])
         )
       })
@@ -36,6 +35,7 @@ export function overview (api: ApiInterfaceRx): () => Observable<DerivedStakingO
       api.derive.session.indexes(),
       api.derive.staking.validators()
     ]).pipe(
+      observeOn(asyncScheduler),
       switchMap(([indexes, { nextElected, validators }]) =>
         combineLatest([
           of({ ...indexes, nextElected, validators }),
@@ -43,9 +43,10 @@ export function overview (api: ApiInterfaceRx): () => Observable<DerivedStakingO
             ? api.query.staking.erasRewardPoints<EraRewardPoints>(indexes.activeEra)
             : api.query.staking.currentEraPointsEarned
               ? retrievePointsPrev(api, indexes.activeEra, nextElected)
-              : of(createType(api.registry, 'EraRewardPoints'))
+              : of(api.registry.createType('EraRewardPoints'))
         ])
       ),
+      observeOn(asyncScheduler),
       map(([info, eraPoints]): DerivedStakingOverview => ({
         ...info, eraPoints
       }))

packages/api-derive/src/staking/query.ts

@@ -7,9 +7,9 @@ import { AccountId, Exposure, Keys, Nominations, RewardDestination, ValidatorPre
 import { ITuple } from '@polkadot/types/types';
 import { DerivedStakingQuery } from '../types';
 
-import { combineLatest, Observable, of } from 'rxjs';
-import { map, switchMap } from 'rxjs/operators';
-import { createType, Option, Vec } from '@polkadot/types';
+import { Observable, asyncScheduler, combineLatest, of } from 'rxjs';
+import { map, observeOn, switchMap } from 'rxjs/operators';
+import { Option, Vec } from '@polkadot/types';
 
 import { memo } from '../util';
 
@@ -80,7 +80,7 @@ function retrieveController (api: ApiInterfaceRx, stashId: AccountId, [queuedKey
  */
 export function query (api: ApiInterfaceRx): (accountId: Uint8Array | string) => Observable<DerivedStakingQuery> {
   return memo((accountId: Uint8Array | string): Observable<DerivedStakingQuery> => {
-    const stashId = createType(api.registry, 'AccountId', accountId);
+    const stashId = api.registry.createType('AccountId', accountId);
 
     return combineLatest([
       api.query.session.queuedKeys<Vec<ITuple<[AccountId, Keys]>>>(),
@@ -88,6 +88,7 @@ export function query (api: ApiInterfaceRx): (accountId: Uint8Array | string) =>
         ? retrieveCurr(api, stashId)
         : retrievePrev(api, stashId)
     ]).pipe(
+      observeOn(asyncScheduler),
       switchMap((result): Observable<DerivedStakingQuery> =>
         retrieveController(api, stashId, result)
       )

packages/api-derive/src/staking/stakerRewards.ts

@@ -7,8 +7,8 @@ import { Balance } from '@polkadot/types/interfaces';
 import { DeriveEraPoints, DeriveEraRewards, DeriveStakerExposure, DeriveStakerReward } from '../types';
 
 import BN from 'bn.js';
-import { Observable, combineLatest } from 'rxjs';
-import { map } from 'rxjs/operators';
+import { Observable, asyncScheduler, combineLatest } from 'rxjs';
+import { map, observeOn } from 'rxjs/operators';
 
 import { memo } from '../util';
 
@@ -67,6 +67,7 @@ export function stakerRewards (api: ApiInterfaceRx): (accountId: Uint8Array | st
       api.derive.staking.erasRewards(startEra),
       api.derive.staking.stakerExposure(stakerId, startEra)
     ]).pipe(
+      observeOn(asyncScheduler),
       map((result): DeriveStakerReward[] =>
         parseRewards(api, stakerId, result)
       )

packages/api-derive/src/staking/validators.ts

@@ -6,8 +6,8 @@ import { ApiInterfaceRx } from '@polkadot/api/types';
 import { AccountId } from '@polkadot/types/interfaces';
 import { DeriveStakingValidators } from '../types';
 
-import { Observable, combineLatest, of } from 'rxjs';
-import { map, switchMap, take } from 'rxjs/operators';
+import { Observable, asyncScheduler, combineLatest, of } from 'rxjs';
+import { map, observeOn, switchMap, take } from 'rxjs/operators';
 import { StorageKey } from '@polkadot/types';
 
 import { memo } from '../util';
@@ -45,6 +45,7 @@ export function validators (api: ApiInterfaceRx): () => Observable<DeriveStaking
         ? api.derive.staking.nextElected()
         : of([])
     ]).pipe(
+      observeOn(asyncScheduler),
       map(([validators, nextElected]): DeriveStakingValidators => ({
         nextElected: nextElected.length
           ? nextElected