diff --git a/packages/api-derive/src/accounts/idAndIndex.ts b/packages/api-derive/src/accounts/idAndIndex.ts index 2722c22550..3b087be5f3 100644 --- a/packages/api-derive/src/accounts/idAndIndex.ts +++ b/packages/api-derive/src/accounts/idAndIndex.ts @@ -7,7 +7,6 @@ import { AccountId, AccountIndex, Address } from '@polkadot/types/interfaces'; import { Observable, of } from 'rxjs'; import { map } from 'rxjs/operators'; import { ApiInterfaceRx } from '@polkadot/api/types'; -import { createType } from '@polkadot/types'; import { isU8a } from '@polkadot/util'; import { decodeAddress } from '@polkadot/util-crypto'; @@ -23,14 +22,14 @@ function retrieve (api: ApiInterfaceRx, address: Address | AccountId | AccountIn : decodeAddress((address || '').toString()); if (decoded.length === 32) { - const accountId = createType(api.registry, 'AccountId', decoded); + const accountId = api.registry.createType('AccountId', decoded); return api.derive.accounts.idToIndex(accountId).pipe( map((accountIndex): AccountIdAndIndex => [accountId, accountIndex]) ); } - const accountIndex = createType(api.registry, 'AccountIndex', decoded); + const accountIndex = api.registry.createType('AccountIndex', decoded); return api.derive.accounts.indexToId(accountIndex).pipe( map((accountId): AccountIdAndIndex => [accountId, accountIndex]) diff --git a/packages/api-derive/src/accounts/indexToId.ts b/packages/api-derive/src/accounts/indexToId.ts index 8d159c2be5..912f9bf7fa 100644 --- a/packages/api-derive/src/accounts/indexToId.ts +++ b/packages/api-derive/src/accounts/indexToId.ts @@ -9,14 +9,14 @@ import { Observable, of } from 'rxjs'; import { map, startWith } from 'rxjs/operators'; import { ApiInterfaceRx } from '@polkadot/api/types'; import { ENUMSET_SIZE } from '@polkadot/types/generic/AccountIndex'; -import { Option, Vec, createType } from '@polkadot/types'; +import { Option, Vec } from '@polkadot/types'; import { memo } from '../util'; function queryEnumSet (api: ApiInterfaceRx, _accountIndex: AccountIndex | string): Observable { const accountIndex = _accountIndex instanceof api.registry.createClass('AccountIndex') ? _accountIndex - : createType(api.registry, 'AccountIndex', _accountIndex); + : api.registry.createType('AccountIndex', _accountIndex); return api.query.indices.enumSet>(accountIndex.div(ENUMSET_SIZE)).pipe( startWith([]), diff --git a/packages/api-derive/src/accounts/indexes.ts b/packages/api-derive/src/accounts/indexes.ts index fe27c17984..3afd97a5c4 100644 --- a/packages/api-derive/src/accounts/indexes.ts +++ b/packages/api-derive/src/accounts/indexes.ts @@ -9,7 +9,7 @@ import { AccountIndexes } from '../types'; import { Observable, of } from 'rxjs'; import { map, startWith, switchMap } from 'rxjs/operators'; import { ENUMSET_SIZE } from '@polkadot/types/generic/AccountIndex'; -import { Vec, createType } from '@polkadot/types'; +import { Vec } from '@polkadot/types'; import { memo } from '../util'; @@ -31,7 +31,7 @@ function queryEnumSet (api: ApiInterfaceRx): Observable { // 64 (0..63 in first) is [1][0] (the first index value in set 2) const index = (outerIndex * enumsetSize) + innerIndex; - indexes[accountId.toString()] = createType(api.registry, 'AccountIndex', index); + indexes[accountId.toString()] = api.registry.createType('AccountIndex', index); }); return indexes; diff --git a/packages/api-derive/src/accounts/info.ts b/packages/api-derive/src/accounts/info.ts index 85b1da5e5a..8f16136695 100644 --- a/packages/api-derive/src/accounts/info.ts +++ b/packages/api-derive/src/accounts/info.ts @@ -7,8 +7,8 @@ import { AccountId, AccountIndex, Address, Balance, Registration } from '@polkad import { ITuple } from '@polkadot/types/types'; import { DeriveAccountInfo, DeriveAccountRegistration } 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 { Bytes, Data, Option, u32 } from '@polkadot/types'; import { u8aToString } from '@polkadot/util'; @@ -86,6 +86,7 @@ function retrieveIdentity (api: ApiInterfaceRx, accountId?: AccountId): Observab ]) : of([undefined, undefined]) ) as Observable<[Option | undefined, Option> | undefined]>).pipe( + observeOn(asyncScheduler), switchMap(([identityOfOpt, superOfOpt]): Observable<[Option | undefined, [AccountId, Data] | undefined]> => { if (identityOfOpt?.isSome) { // this identity has something set @@ -103,6 +104,7 @@ function retrieveIdentity (api: ApiInterfaceRx, accountId?: AccountId): Observab // nothing of value returned return of([undefined, undefined]); }), + observeOn(asyncScheduler), map(([identityOfOpt, superOf]): DeriveAccountRegistration => extractIdentity(identityOfOpt, superOf) ) @@ -116,6 +118,7 @@ function retrieveIdentity (api: ApiInterfaceRx, accountId?: AccountId): Observab export function info (api: ApiInterfaceRx): (address?: AccountIndex | AccountId | Address | string | null) => Observable { return memo((address?: AccountIndex | AccountId | Address | string | null): Observable => api.derive.accounts.idAndIndex(address).pipe( + observeOn(asyncScheduler), switchMap(([accountId, accountIndex]): Observable<[Partial, DeriveAccountRegistration, string?]> => combineLatest([ of({ accountId, accountIndex }), @@ -123,6 +126,7 @@ export function info (api: ApiInterfaceRx): (address?: AccountIndex | AccountId retrieveNick(api, accountId) ]) ), + observeOn(asyncScheduler), map(([{ accountId, accountIndex }, identity, nickname]): DeriveAccountInfo => ({ accountId, accountIndex, identity, nickname })) diff --git a/packages/api-derive/src/balances/account.ts b/packages/api-derive/src/balances/account.ts index f43fba0eb7..023237e0c6 100644 --- a/packages/api-derive/src/balances/account.ts +++ b/packages/api-derive/src/balances/account.ts @@ -6,10 +6,9 @@ import { AccountId, AccountData, AccountIndex, AccountInfo, Address, Balance, In import { ITuple } from '@polkadot/types/types'; import { DerivedBalancesAccount } from '../types'; -import { combineLatest, of, Observable } from 'rxjs'; -import { map, switchMap } from 'rxjs/operators'; +import { Observable, asyncScheduler, combineLatest, of } from 'rxjs'; +import { map, observeOn, switchMap } from 'rxjs/operators'; import { ApiInterfaceRx } from '@polkadot/api/types'; -import { createType } from '@polkadot/types'; import { memo } from '../util'; @@ -23,7 +22,7 @@ function calcBalances (api: ApiInterfaceRx, [accountId, [freeBalance, reservedBa frozenFee, frozenMisc, reservedBalance, - votingBalance: createType(api.registry, 'Balance', freeBalance.add(reservedBalance)) + votingBalance: api.registry.createType('Balance', freeBalance.add(reservedBalance)) }; } @@ -35,7 +34,7 @@ function queryBalancesFree (api: ApiInterfaceRx, accountId: AccountId): Observab [api.query.system.accountNonce, accountId] ]).pipe( map(([freeBalance, reservedBalance, accountNonce]): Result => - [freeBalance, reservedBalance, createType(api.registry, 'Balance'), createType(api.registry, 'Balance'), accountNonce] + [freeBalance, reservedBalance, api.registry.createType('Balance'), api.registry.createType('Balance'), accountNonce] ) ); } @@ -83,6 +82,7 @@ function queryCurrent (api: ApiInterfaceRx, accountId: AccountId): Observable Observable { return memo((address: AccountIndex | AccountId | Address | string): Observable => api.derive.accounts.info(address).pipe( + observeOn(asyncScheduler), switchMap(({ accountId }): Observable<[AccountId, Result]> => (accountId ? combineLatest([ @@ -93,9 +93,10 @@ export function account (api: ApiInterfaceRx): (address: AccountIndex | AccountI ? queryBalancesAccount(api, accountId) : queryBalancesFree(api, accountId) ]) - : of([createType(api.registry, 'AccountId'), [createType(api.registry, 'Balance'), createType(api.registry, 'Balance'), createType(api.registry, 'Balance'), createType(api.registry, 'Balance'), createType(api.registry, 'Index')]]) + : of([api.registry.createType('AccountId'), [api.registry.createType('Balance'), api.registry.createType('Balance'), api.registry.createType('Balance'), api.registry.createType('Balance'), api.registry.createType('Index')]]) ) ), + observeOn(asyncScheduler), map((result): DerivedBalancesAccount => calcBalances(api, result)) )); } diff --git a/packages/api-derive/src/balances/all.ts b/packages/api-derive/src/balances/all.ts index fa2e87d7f7..1c55df689a 100644 --- a/packages/api-derive/src/balances/all.ts +++ b/packages/api-derive/src/balances/all.ts @@ -6,10 +6,10 @@ import { AccountId, AccountIndex, Address, Balance, BalanceLock, BalanceLockTo21 import { DerivedBalancesAccount, DerivedBalancesAll } from '../types'; import BN from 'bn.js'; -import { combineLatest, of, Observable } from 'rxjs'; -import { map, switchMap } from 'rxjs/operators'; +import { Observable, asyncScheduler, combineLatest, of } from 'rxjs'; +import { map, observeOn, switchMap } from 'rxjs/operators'; import { ApiInterfaceRx } from '@polkadot/api/types'; -import { Option, Vec, createType } from '@polkadot/types'; +import { Option, Vec } from '@polkadot/types'; import { bnMax } from '@polkadot/util'; import { memo } from '../util'; @@ -18,7 +18,7 @@ type ResultBalance = [VestingInfo | null, (BalanceLock | BalanceLockTo212)[]]; type Result = [DerivedBalancesAccount, BlockNumber, ResultBalance]; function calcBalances (api: ApiInterfaceRx, [{ accountId, accountNonce, freeBalance, frozenFee, frozenMisc, reservedBalance, votingBalance }, bestNumber, [vesting, locks]]: Result): DerivedBalancesAll { - let lockedBalance = createType(api.registry, 'Balance'); + let lockedBalance = api.registry.createType('Balance'); let lockedBreakdown: (BalanceLock | BalanceLockTo212)[] = []; let allLocked = false; @@ -32,16 +32,16 @@ function calcBalances (api: ApiInterfaceRx, [{ accountId, accountNonce, freeBala // get the maximum of the locks according to https://github.com/paritytech/substrate/blob/master/srml/balances/src/lib.rs#L699 if (notAll.length) { - lockedBalance = createType(api.registry, 'Balance', bnMax(...notAll.map(({ amount }): Balance => amount))); + lockedBalance = api.registry.createType('Balance', bnMax(...notAll.map(({ amount }): Balance => amount))); } } // Calculate the vesting balances, // - offset = balance locked at startingBlock // - perBlock is the unlock amount - const { locked: vestingTotal, perBlock, startingBlock } = vesting || createType(api.registry, 'VestingInfo'); + const { locked: vestingTotal, perBlock, startingBlock } = vesting || api.registry.createType('VestingInfo'); const isStarted = bestNumber.gt(startingBlock); - const vestedBalance = createType(api.registry, 'Balance', isStarted ? perBlock.mul(bestNumber.sub(startingBlock)) : 0); + const vestedBalance = api.registry.createType('Balance', isStarted ? perBlock.mul(bestNumber.sub(startingBlock)) : 0); const isVesting = isStarted && vestedBalance.lt(vestingTotal); // The available balance & vested has an interplay here @@ -53,7 +53,7 @@ function calcBalances (api: ApiInterfaceRx, [{ accountId, accountNonce, freeBala // "" const floating = freeBalance.sub(lockedBalance); const extraReceived = isVesting ? freeBalance.sub(vestingTotal) : new BN(0); - const availableBalance = createType(api.registry, 'Balance', allLocked ? 0 : bnMax(new BN(0), isVesting && floating.gt(vestedBalance) ? vestedBalance.add(extraReceived) : floating)); + const availableBalance = api.registry.createType('Balance', allLocked ? 0 : bnMax(new BN(0), isVesting && floating.gt(vestedBalance) ? vestedBalance.add(extraReceived) : floating)); return { accountId, @@ -84,7 +84,7 @@ function queryOld (api: ApiInterfaceRx, accountId: AccountId): Observable, Option] => - [locks, createType(api.registry, 'Option')] + [locks, api.registry.createType('Option')] ) ) ).pipe( @@ -130,6 +130,7 @@ function queryCurrent (api: ApiInterfaceRx, accountId: AccountId): Observable Observable { return memo((address: AccountIndex | AccountId | Address | string): Observable => api.derive.balances.account(address).pipe( + observeOn(asyncScheduler), switchMap((account): Observable => (!account.accountId.isEmpty ? combineLatest([ @@ -139,9 +140,10 @@ export function all (api: ApiInterfaceRx): (address: AccountIndex | AccountId | ? queryCurrent(api, account.accountId) : queryOld(api, account.accountId) ]) - : of([account, createType(api.registry, 'BlockNumber'), [null, createType(api.registry, 'Vec')]]) + : of([account, api.registry.createType('BlockNumber'), [null, api.registry.createType('Vec')]]) ) ), + observeOn(asyncScheduler), map((result): DerivedBalancesAll => calcBalances(api, result)) )); } diff --git a/packages/api-derive/src/balances/fees.ts b/packages/api-derive/src/balances/fees.ts index 640b5f6460..5335375ae0 100644 --- a/packages/api-derive/src/balances/fees.ts +++ b/packages/api-derive/src/balances/fees.ts @@ -8,7 +8,6 @@ import { DerivedFees } from '../types'; import { Observable, of } from 'rxjs'; import { map } from 'rxjs/operators'; -import { createType } from '@polkadot/types'; import { memo } from '../util'; @@ -29,13 +28,13 @@ export function fees (api: ApiInterfaceRx): () => Observable { return memo((): Observable => of([ // deprecated - remove - (api.consts.balances?.creationFee as Balance) || createType(api.registry, 'Balance'), - (api.consts.balances?.transferFee as Balance) || createType(api.registry, 'Balance'), + (api.consts.balances?.creationFee as Balance) || api.registry.createType('Balance'), + (api.consts.balances?.transferFee as Balance) || api.registry.createType('Balance'), // current - (api.consts.balances?.existentialDeposit as Balance) || createType(api.registry, 'Balance'), - (api.consts.transactionPayment?.transactionBaseFee as Balance) || createType(api.registry, 'Balance'), - (api.consts.transactionPayment?.transactionByteFee as Balance) || createType(api.registry, 'Balance') + (api.consts.balances?.existentialDeposit as Balance) || api.registry.createType('Balance'), + (api.consts.transactionPayment?.transactionBaseFee as Balance) || api.registry.createType('Balance'), + (api.consts.transactionPayment?.transactionByteFee as Balance) || api.registry.createType('Balance') ]).pipe( map(([creationFee, transferFee, existentialDeposit, transactionBaseFee, transactionByteFee]): DerivedFees => ({ creationFee, diff --git a/packages/api-derive/src/balances/votingBalances.ts b/packages/api-derive/src/balances/votingBalances.ts index af95c210a0..eef10f828f 100644 --- a/packages/api-derive/src/balances/votingBalances.ts +++ b/packages/api-derive/src/balances/votingBalances.ts @@ -5,7 +5,8 @@ import { AccountId, AccountIndex, Address } from '@polkadot/types/interfaces'; import { DerivedBalancesAccount } from '../types'; -import { combineLatest, Observable, of } from 'rxjs'; +import { Observable, asyncScheduler, combineLatest, of } from 'rxjs'; +import { observeOn } from 'rxjs/operators'; import { ApiInterfaceRx } from '@polkadot/api/types'; import { memo } from '../util'; @@ -16,7 +17,8 @@ export function votingBalances (api: ApiInterfaceRx): (addresses?: (AccountId | ? of([] as DerivedBalancesAccount[]) : combineLatest( addresses.map((accountId): Observable => - api.derive.balances.account(accountId)) - ) + api.derive.balances.account(accountId) + ) + ).pipe(observeOn(asyncScheduler)) ); } diff --git a/packages/api-derive/src/chain/bestNumberLag.ts b/packages/api-derive/src/chain/bestNumberLag.ts index 134b1d7dba..9453d6057c 100644 --- a/packages/api-derive/src/chain/bestNumberLag.ts +++ b/packages/api-derive/src/chain/bestNumberLag.ts @@ -4,10 +4,9 @@ import { BlockNumber } from '@polkadot/types/interfaces'; -import { Observable, combineLatest } from 'rxjs'; -import { map } from 'rxjs/operators'; +import { Observable, asyncScheduler, combineLatest } from 'rxjs'; +import { map, observeOn } from 'rxjs/operators'; import { ApiInterfaceRx } from '@polkadot/api/types'; -import { createType } from '@polkadot/types'; import { memo } from '../util'; @@ -30,8 +29,9 @@ export function bestNumberLag (api: ApiInterfaceRx): () => Observable - createType(api.registry, 'BlockNumber', bestNumber.sub(bestNumberFinalized)) + api.registry.createType('BlockNumber', bestNumber.sub(bestNumberFinalized)) ) )); } diff --git a/packages/api-derive/src/chain/getHeader.ts b/packages/api-derive/src/chain/getHeader.ts index 7f11cb2779..5de051fc37 100644 --- a/packages/api-derive/src/chain/getHeader.ts +++ b/packages/api-derive/src/chain/getHeader.ts @@ -4,8 +4,8 @@ import { ApiInterfaceRx } from '@polkadot/api/types'; -import { Observable, combineLatest, of } from 'rxjs'; -import { catchError, map } from 'rxjs/operators'; +import { Observable, asyncScheduler, combineLatest, of } from 'rxjs'; +import { catchError, map, observeOn } from 'rxjs/operators'; import { HeaderExtended } from '../type'; import { memo } from '../util'; @@ -32,6 +32,7 @@ export function getHeader (api: ApiInterfaceRx): (hash: Uint8Array | string) => ? api.query.session.validators.at(hash) : of([]) ]).pipe( + observeOn(asyncScheduler), map(([header, validators]): HeaderExtended => new HeaderExtended(api.registry, header, validators) ), diff --git a/packages/api-derive/src/chain/subscribeNewHeads.ts b/packages/api-derive/src/chain/subscribeNewHeads.ts index d895bf38d4..6061b0cf10 100644 --- a/packages/api-derive/src/chain/subscribeNewHeads.ts +++ b/packages/api-derive/src/chain/subscribeNewHeads.ts @@ -4,8 +4,8 @@ import { ApiInterfaceRx } from '@polkadot/api/types'; -import { Observable, combineLatest, of } from 'rxjs'; -import { map } from 'rxjs/operators'; +import { Observable, asyncScheduler, combineLatest, of } from 'rxjs'; +import { map, observeOn } from 'rxjs/operators'; import { HeaderExtended } from '../type'; import { memo } from '../util'; @@ -31,6 +31,7 @@ export function subscribeNewHeads (api: ApiInterfaceRx): () => Observable
new HeaderExtended(api.registry, header, validators) ) diff --git a/packages/api-derive/src/collective/proposals.ts b/packages/api-derive/src/collective/proposals.ts index 5a84eb816e..5bc9f04e22 100644 --- a/packages/api-derive/src/collective/proposals.ts +++ b/packages/api-derive/src/collective/proposals.ts @@ -6,8 +6,8 @@ import { ApiInterfaceRx } from '@polkadot/api/types'; import { Hash, Proposal, Votes } from '@polkadot/types/interfaces'; import { DerivedCollectiveProposal } from '../types'; -import { combineLatest, Observable, 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'; type Result = [Hash[], Option[], Option[]]; @@ -30,6 +30,7 @@ export function proposals (api: ApiInterfaceRx, section: 'council' | 'technicalC return (): Observable => api.query[section]?.proposals ? api.query[section].proposals().pipe( + observeOn(asyncScheduler), switchMap((hashes: Hash[]): Observable => hashes.length ? combineLatest([ @@ -39,6 +40,7 @@ export function proposals (api: ApiInterfaceRx, section: 'council' | 'technicalC ]) : of([[], [], []]) ), + observeOn(asyncScheduler), map(parse) ) : of([] as DerivedCollectiveProposal[]); diff --git a/packages/api-derive/src/contracts/fees.ts b/packages/api-derive/src/contracts/fees.ts index 7d5050bba2..115aa667fb 100644 --- a/packages/api-derive/src/contracts/fees.ts +++ b/packages/api-derive/src/contracts/fees.ts @@ -8,7 +8,6 @@ import { DerivedContractFees } from '../types'; import BN from 'bn.js'; import { Observable, of } from 'rxjs'; import { map } from 'rxjs/operators'; -import { createType } from '@polkadot/types'; import { memo } from '../util'; @@ -18,8 +17,8 @@ type ResultV2 = [BN, BN, BN, BN, BN, BN, BN, BN, BN]; function queryConstants (api: ApiInterfaceRx): Observable { return of([ // deprecated - api.consts.contracts.creationFee || createType(api.registry, 'Balance'), - api.consts.contracts.transferFee || createType(api.registry, 'Balance'), + api.consts.contracts.creationFee || api.registry.createType('Balance'), + api.consts.contracts.transferFee || api.registry.createType('Balance'), // current api.consts.contracts.callBaseFee, diff --git a/packages/api-derive/src/democracy/proposals.ts b/packages/api-derive/src/democracy/proposals.ts index 3c77cf5c37..68c51a1004 100644 --- a/packages/api-derive/src/democracy/proposals.ts +++ b/packages/api-derive/src/democracy/proposals.ts @@ -6,10 +6,10 @@ import { AccountId, Balance, BlockNumber, Hash, PropIndex, Proposal } from '@pol import { ITuple } from '@polkadot/types/types'; import { DeriveProposal } 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 { ApiInterfaceRx } from '@polkadot/api/types'; -import { Bytes, Option, Vec, createType } from '@polkadot/types'; +import { Bytes, Option, Vec } from '@polkadot/types'; import { memo } from '../util'; @@ -37,7 +37,7 @@ function parse (api: ApiInterfaceRx, { depositors, proposals, preimages }: Resul // we could end up in a situation where the proposal is non-decodable, e.g. after an upgrade if (preimage) { try { - proposal = createType(api.registry, 'Proposal', preimage[0].toU8a(true)); + proposal = api.registry.createType('Proposal', preimage[0].toU8a(true)); } catch (error) { console.error(error); } @@ -65,6 +65,7 @@ export function proposals (api: ApiInterfaceRx): () => Observable => api.query.democracy?.publicProps && api.query.democracy?.preimages ? api.query.democracy.publicProps().pipe( + observeOn(asyncScheduler), switchMap((proposals) => combineLatest([ of(proposals), @@ -74,6 +75,7 @@ export function proposals (api: ApiInterfaceRx): () => Observable index)) ]) ), + observeOn(asyncScheduler), map(([proposals, preimages, depositors]): DeriveProposal[] => parse(api, { depositors, proposals, preimages }) ) diff --git a/packages/api-derive/src/democracy/referendumInfo.ts b/packages/api-derive/src/democracy/referendumInfo.ts index 33c4532706..ed0eb0abd1 100644 --- a/packages/api-derive/src/democracy/referendumInfo.ts +++ b/packages/api-derive/src/democracy/referendumInfo.ts @@ -10,7 +10,7 @@ import BN from 'bn.js'; import { Observable, of } from 'rxjs'; import { map, switchMap } from 'rxjs/operators'; import { ApiInterfaceRx } from '@polkadot/api/types'; -import { Option, createType } from '@polkadot/types'; +import { Option } from '@polkadot/types'; import { memo } from '../util'; @@ -25,11 +25,11 @@ function constructInfo (api: ApiInterfaceRx, index: BN | number, _info: Option[]) : api.query.democracy.referendumInfoOf.multi>(ids) ).pipe( + observeOn(asyncScheduler), switchMap((infos): Observable<(DerivedReferendum | null)[]> => combineLatest( - ...ids.map((id, index): Observable => + ids.map((id, index): Observable => retrieveInfo(api, id, infos[index]) ) ) ), + observeOn(asyncScheduler), map((infos): DerivedReferendum[] => infos.filter((referendum): referendum is DerivedReferendum => !!referendum) ) diff --git a/packages/api-derive/src/democracy/referendumVotesFor.ts b/packages/api-derive/src/democracy/referendumVotesFor.ts index 1e2d88a1da..970196c88e 100644 --- a/packages/api-derive/src/democracy/referendumVotesFor.ts +++ b/packages/api-derive/src/democracy/referendumVotesFor.ts @@ -5,10 +5,10 @@ import { AccountId, Vote } from '@polkadot/types/interfaces'; import BN from 'bn.js'; -import { combineLatest, Observable, of } from 'rxjs'; -import { map, switchMap } from 'rxjs/operators'; +import { Observable, asyncScheduler, combineLatest, of } from 'rxjs'; +import { map, observeOn, switchMap } from 'rxjs/operators'; import { ApiInterfaceRx } from '@polkadot/api/types'; -import { Vec, createType } from '@polkadot/types'; +import { Vec } from '@polkadot/types'; import { DerivedBalancesAccount, DerivedReferendumVote } from '../types'; import { memo } from '../util'; @@ -16,6 +16,7 @@ import { memo } from '../util'; export function referendumVotesFor (api: ApiInterfaceRx): (referendumId: BN | number) => Observable { return memo((referendumId: BN | number): Observable => api.query.democracy.votersFor>(referendumId).pipe( + observeOn(asyncScheduler), switchMap((votersFor): Observable<[Vec, Vote[], DerivedBalancesAccount[]]> => combineLatest([ of(votersFor), @@ -23,11 +24,12 @@ export function referendumVotesFor (api: ApiInterfaceRx): (referendumId: BN | nu api.derive.balances.votingBalances(votersFor) ]) ), + observeOn(asyncScheduler), map(([votersFor, votes, balances]): DerivedReferendumVote[] => votersFor.map((accountId, index): DerivedReferendumVote => ({ accountId, - balance: balances[index].votingBalance || createType(api.registry, 'Balance'), - vote: votes[index] || createType(api.registry, 'Vote') + balance: balances[index].votingBalance || api.registry.createType('Balance'), + vote: votes[index] || api.registry.createType('Vote') } as DerivedReferendumVote)) ) )); diff --git a/packages/api-derive/src/elections/info.ts b/packages/api-derive/src/elections/info.ts index 6f68befc29..bffdff3f1e 100644 --- a/packages/api-derive/src/elections/info.ts +++ b/packages/api-derive/src/elections/info.ts @@ -8,7 +8,7 @@ import { ITuple } from '@polkadot/types/types'; import { Observable } from 'rxjs'; import { map } from 'rxjs/operators'; import { ApiInterfaceRx } from '@polkadot/api/types'; -import { createType, Vec, u32 } from '@polkadot/types'; +import { Vec, u32 } from '@polkadot/types'; import { DerivedElectionsInfo } from '../types'; import { memo } from '../util'; @@ -28,12 +28,12 @@ function queryElections (api: ApiInterfaceRx): Observable ]).pipe( map(([councilMembers, candidates, members, runnersUp]): DerivedElectionsInfo => ({ candidates, - candidateCount: createType(api.registry, 'u32', candidates.length), + candidateCount: api.registry.createType('u32', candidates.length), candidacyBond: api.consts[section].candidacyBond as Balance, desiredSeats: api.consts[section].desiredMembers as u32, members: members.length ? members.sort(sortAccounts) - : councilMembers.map((accountId): [AccountId, Balance] => [accountId, createType(api.registry, 'Balance')]), + : councilMembers.map((accountId): [AccountId, Balance] => [accountId, api.registry.createType('Balance')]), runnersUp: runnersUp.sort(sortAccounts), termDuration: api.consts[section].termDuration as BlockNumber, votingBond: api.consts[section].votingBond as Balance diff --git a/packages/api-derive/src/imOnline/receivedHeartbeats.ts b/packages/api-derive/src/imOnline/receivedHeartbeats.ts index d73a0b30d0..5180e47c35 100644 --- a/packages/api-derive/src/imOnline/receivedHeartbeats.ts +++ b/packages/api-derive/src/imOnline/receivedHeartbeats.ts @@ -5,8 +5,8 @@ import { AccountId } from '@polkadot/types/interfaces'; import { DerivedHeartbeats } from '../types'; -import { of, Observable, combineLatest } from 'rxjs'; -import { map, switchMap } from 'rxjs/operators'; +import { Observable, asyncScheduler, of, combineLatest } from 'rxjs'; +import { map, observeOn, switchMap } from 'rxjs/operators'; import { ApiInterfaceRx } from '@polkadot/api/types'; import { Bytes, Option, u32 } from '@polkadot/types'; @@ -19,6 +19,7 @@ export function receivedHeartbeats (api: ApiInterfaceRx): () => Observable => api.query.imOnline?.receivedHeartbeats ? api.derive.staking.overview().pipe( + observeOn(asyncScheduler), switchMap(({ currentIndex, validators }): Observable<[AccountId[], Option[], u32[]]> => combineLatest([ of(validators), @@ -28,6 +29,7 @@ export function receivedHeartbeats (api: ApiInterfaceRx): () => Observable [currentIndex, address])) ]) ), + observeOn(asyncScheduler), map(([validators, heartbeats, numBlocks]): DerivedHeartbeats => validators.reduce((result: DerivedHeartbeats, validator, index): DerivedHeartbeats => ({ ...result, diff --git a/packages/api-derive/src/parachains/info.ts b/packages/api-derive/src/parachains/info.ts index 5c530f6006..57076ef5f4 100644 --- a/packages/api-derive/src/parachains/info.ts +++ b/packages/api-derive/src/parachains/info.ts @@ -9,7 +9,6 @@ import { Active, DidUpdate, Heads, ParaInfoResult, PendingSwap, RelayDispatchQue import { Observable, of } from 'rxjs'; import { map } from 'rxjs/operators'; import { ApiInterfaceRx } from '@polkadot/api/types'; -import { createType } from '@polkadot/types'; import { memo } from '../util'; @@ -92,7 +91,7 @@ export function info (api: ApiInterfaceRx): (id: ParaId | number) => Observable< ]) .pipe( map((result: Result): DeriveParachainFull | null => - parse(createType(api.registry, 'ParaId', id), result) + parse(api.registry.createType('ParaId', id), result) ) ) : of(null) diff --git a/packages/api-derive/src/parachains/overview.ts b/packages/api-derive/src/parachains/overview.ts index e3a32a2a72..9a652eb211 100644 --- a/packages/api-derive/src/parachains/overview.ts +++ b/packages/api-derive/src/parachains/overview.ts @@ -6,8 +6,8 @@ import { ParaId } from '@polkadot/types/interfaces'; import { DeriveParachain, DeriveParachainInfo } from '../types'; import { DidUpdate, ParaInfoResult, PendingSwap, RelayDispatchQueueSize } 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 { ApiInterfaceRx } from '@polkadot/api/types'; import { memo } from '../util'; @@ -38,6 +38,7 @@ export function overview (api: ApiInterfaceRx): () => Observable => api.query.registrar?.parachains && api.query.parachains ? api.query.registrar.parachains().pipe( + observeOn(asyncScheduler), switchMap((paraIds) => combineLatest([ of(paraIds), @@ -47,6 +48,7 @@ export function overview (api: ApiInterfaceRx): () => Observable(paraIds) ]) ), + observeOn(asyncScheduler), map((result: Result): DeriveParachain[] => parse(result) ) diff --git a/packages/api-derive/src/session/indexes.ts b/packages/api-derive/src/session/indexes.ts index 0e7c519359..3a8c3ef204 100644 --- a/packages/api-derive/src/session/indexes.ts +++ b/packages/api-derive/src/session/indexes.ts @@ -8,7 +8,7 @@ import { DeriveSessionIndexes } from '../types'; import { Observable, of } from 'rxjs'; import { map } from 'rxjs/operators'; -import { createType, Option, u32 } from '@polkadot/types'; +import { Option, u32 } from '@polkadot/types'; import { memo } from '../util'; @@ -34,7 +34,7 @@ function queryNoActive (api: ApiInterfaceRx): Observable { ]).pipe( map(([currentEra, currentIndex, validatorCount]): Result => [ currentEra, - createType(api.registry, 'Option'), + api.registry.createType('Option'), currentEra, currentIndex, validatorCount @@ -56,7 +56,7 @@ function query (api: ApiInterfaceRx): Observable { return [ activeEra, activeEraStart, - currentEra.unwrapOr(createType(api.registry, 'EraIndex')), + currentEra.unwrapOr(api.registry.createType('EraIndex')), currentIndex, validatorCount ]; @@ -67,11 +67,11 @@ function query (api: ApiInterfaceRx): Observable { // empty set when none is available function empty (api: ApiInterfaceRx): Observable { return of([ - createType(api.registry, 'EraIndex'), - createType(api.registry, 'Option'), - createType(api.registry, 'EraIndex'), - createType(api.registry, 'SessionIndex', 1), - createType(api.registry, 'u32') + api.registry.createType('EraIndex'), + api.registry.createType('Option'), + api.registry.createType('EraIndex'), + api.registry.createType('SessionIndex', 1), + api.registry.createType('u32') ]); } diff --git a/packages/api-derive/src/session/info.ts b/packages/api-derive/src/session/info.ts index 2f26b2c41b..98081f67ef 100644 --- a/packages/api-derive/src/session/info.ts +++ b/packages/api-derive/src/session/info.ts @@ -5,10 +5,10 @@ import { SessionIndex } from '@polkadot/types/interfaces'; import { DerivedSessionInfo, DeriveSessionIndexes } 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 { ApiInterfaceRx } from '@polkadot/api/types'; -import { createType, Option, u64 } from '@polkadot/types'; +import { Option, u64 } from '@polkadot/types'; import { memo } from '../util'; @@ -27,12 +27,12 @@ function createDerived (api: ApiInterfaceRx, [[hasBabe, epochDuration, sessionsP activeEraStart, currentEra, currentIndex, - eraLength: createType(api.registry, 'BlockNumber', sessionsPerEra.mul(epochDuration)), - eraProgress: createType(api.registry, 'BlockNumber', eraProgress), + eraLength: api.registry.createType('BlockNumber', sessionsPerEra.mul(epochDuration)), + eraProgress: api.registry.createType('BlockNumber', eraProgress), isEpoch: hasBabe, sessionLength: epochDuration, sessionsPerEra, - sessionProgress: createType(api.registry, 'BlockNumber', sessionProgress), + sessionProgress: api.registry.createType('BlockNumber', sessionProgress), validatorCount }; } @@ -43,15 +43,15 @@ function queryAura (api: ApiInterfaceRx): Observable { createDerived(api, [ [ false, - createType(api.registry, 'u64', 1), - api.consts.staking?.sessionsPerEra || createType(api.registry, 'SessionIndex', 1) + api.registry.createType('u64', 1), + api.consts.staking?.sessionsPerEra || api.registry.createType('SessionIndex', 1) ], indexes, [ - createType(api.registry, 'u64', 1), - createType(api.registry, 'u64', 1), - createType(api.registry, 'u64', 1), - createType(api.registry, 'SessionIndex', 1) + api.registry.createType('u64', 1), + api.registry.createType('u64', 1), + api.registry.createType('u64', 1), + api.registry.createType('SessionIndex', 1) ] ]) ) @@ -60,6 +60,7 @@ function queryAura (api: ApiInterfaceRx): Observable { function queryBabe (api: ApiInterfaceRx): Observable<[DeriveSessionIndexes, ResultSlotsFlat]> { return api.derive.session.indexes().pipe( + observeOn(asyncScheduler), switchMap((indexes): Observable<[DeriveSessionIndexes, ResultSlots]> => combineLatest([ of(indexes), @@ -71,8 +72,9 @@ function queryBabe (api: ApiInterfaceRx): Observable<[DeriveSessionIndexes, Resu ]) ]) ), + observeOn(asyncScheduler), map(([indexes, [currentSlot, epochIndex, genesisSlot, optStartIndex]]): [DeriveSessionIndexes, ResultSlotsFlat] => [ - indexes, [currentSlot, epochIndex, genesisSlot, optStartIndex.unwrapOr(createType(api.registry, 'SessionIndex', 1))] + indexes, [currentSlot, epochIndex, genesisSlot, optStartIndex.unwrapOr(api.registry.createType('SessionIndex', 1))] ]) ); } @@ -99,6 +101,7 @@ export function info (api: ApiInterfaceRx): () => Observable ? queryBabe(api) // 2.x with Babe : queryBabeNoHistory(api) ).pipe( + observeOn(asyncScheduler), map(([indexes, slots]: [DeriveSessionIndexes, ResultSlotsFlat]): DerivedSessionInfo => createDerived(api, [ [true, api.consts.babe.epochDuration, api.consts.staking.sessionsPerEra], diff --git a/packages/api-derive/src/society/candidates.ts b/packages/api-derive/src/society/candidates.ts index 6860ac7289..3835540d70 100644 --- a/packages/api-derive/src/society/candidates.ts +++ b/packages/api-derive/src/society/candidates.ts @@ -7,8 +7,8 @@ import { AccountId, BalanceOf, Bid, BidKind } from '@polkadot/types/interfaces'; import { ITuple } from '@polkadot/types/types'; import { DeriveSocietyCandidate } from '../types'; -import { combineLatest, of, Observable } from 'rxjs'; -import { map, switchMap } from 'rxjs/operators'; +import { Observable, asyncScheduler, combineLatest, of } from 'rxjs'; +import { map, observeOn, switchMap } from 'rxjs/operators'; import { Option, Vec } from '@polkadot/types'; import { memo } from '../util'; @@ -22,6 +22,7 @@ type Result = [Bid[], ResultSuspend[]] export function candidates (api: ApiInterfaceRx): () => Observable { return memo((): Observable => api.query.society.candidates>().pipe( + observeOn(asyncScheduler), switchMap((candidates: Vec): Observable => combineLatest([ of(candidates), @@ -29,6 +30,7 @@ export function candidates (api: ApiInterfaceRx): () => Observable who)) ]) ), + observeOn(asyncScheduler), map(([candidates, suspended]: Result): DeriveSocietyCandidate[] => candidates.map(({ who, kind, value }, index) => ({ accountId: who, diff --git a/packages/api-derive/src/society/members.ts b/packages/api-derive/src/society/members.ts index 0728d6ed56..9aa4442b6e 100644 --- a/packages/api-derive/src/society/members.ts +++ b/packages/api-derive/src/society/members.ts @@ -6,8 +6,8 @@ import { ApiInterfaceRx } from '@polkadot/api/types'; import { AccountId } from '@polkadot/types/interfaces'; import { DeriveSocietyMember } from '../types'; -import { combineLatest, Observable } from 'rxjs'; -import { switchMap } from 'rxjs/operators'; +import { Observable, asyncScheduler, combineLatest } from 'rxjs'; +import { observeOn, switchMap } from 'rxjs/operators'; import { memo } from '../util'; @@ -17,10 +17,13 @@ import { memo } from '../util'; export function members (api: ApiInterfaceRx): () => Observable { return memo((): Observable => api.query.society.members().pipe( + observeOn(asyncScheduler), switchMap((members): Observable => - combineLatest(members.map((accountId): Observable => - api.derive.society.member(accountId) - )) + combineLatest( + members.map((accountId): Observable => + api.derive.society.member(accountId) + ) + ) ) ) ); diff --git a/packages/api-derive/src/staking/account.ts b/packages/api-derive/src/staking/account.ts index e08210fe91..63323d0068 100644 --- a/packages/api-derive/src/staking/account.ts +++ b/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)) )); diff --git a/packages/api-derive/src/staking/controllers.ts b/packages/api-derive/src/staking/controllers.ts index a65a2d2199..607101f876 100644 --- a/packages/api-derive/src/staking/controllers.ts +++ b/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[]]; export function controllers (api: ApiInterfaceRx): () => Observable { return memo((): Observable<[AccountId[], Option[]]> => api.query.staking.validators, Vec]>>().pipe( + observeOn(asyncScheduler), switchMap(([stashIds]): Observable => combineLatest([ of(stashIds), @@ -28,6 +29,7 @@ export function controllers (api: ApiInterfaceRx): () => Observable>(stashIds) ]) - ) + ), + observeOn(asyncScheduler) )); } diff --git a/packages/api-derive/src/staking/electedInfo.ts b/packages/api-derive/src/staking/electedInfo.ts index 5fcf92eb5a..55fa2ece26 100644 --- a/packages/api-derive/src/staking/electedInfo.ts +++ b/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 { return memo((): Observable => api.derive.staking.validators().pipe( + observeOn(asyncScheduler), switchMap(({ nextElected }): Observable<[AccountId[], DerivedStakingQuery[]]> => combineLatest([ of(nextElected), - combineLatest(nextElected.map((accountId): Observable => - api.derive.staking.query(accountId) - )) + combineLatest( + nextElected.map((accountId): Observable => + api.derive.staking.query(accountId) + ) + ) ]) ), + observeOn(asyncScheduler), map(([nextElected, info]): DerivedStakingElected => ({ info, nextElected diff --git a/packages/api-derive/src/staking/erasExposure.ts b/packages/api-derive/src/staking/erasExposure.ts index 5a80d4a29a..682cd52a50 100644 --- a/packages/api-derive/src/staking/erasExposure.ts +++ b/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 { return memo((era: EraIndex): Observable => 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 { return memo((withActive?: boolean | BN | number): Observable => api.derive.staking.erasHistoric(withActive).pipe( + observeOn(asyncScheduler), switchMap((eras): Observable => eras.length - ? combineLatest(eras.map((era) => api.derive.staking.eraExposure(era))) + ? combineLatest( + eras.map((era) => api.derive.staking.eraExposure(era)) + ) : of([]) ) ) diff --git a/packages/api-derive/src/staking/erasPoints.ts b/packages/api-derive/src/staking/erasPoints.ts index 7b3948b63c..066b0230e1 100644 --- a/packages/api-derive/src/staking/erasPoints.ts +++ b/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 { return memo((withActive?: boolean | number): Observable => 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, diff --git a/packages/api-derive/src/staking/erasRewards.ts b/packages/api-derive/src/staking/erasRewards.ts index d4df51b332..8d747ad27c 100644 --- a/packages/api-derive/src/staking/erasRewards.ts +++ b/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 { return memo((withActive?: boolean | BN | number): Observable => api.derive.staking.erasHistoric(withActive).pipe( + observeOn(asyncScheduler), switchMap((eras): Observable<[EraIndex[], Option[]]> => 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, diff --git a/packages/api-derive/src/staking/overview.ts b/packages/api-derive/src/staking/overview.ts index 7b9f10e529..cb0fabb527 100644 --- a/packages/api-derive/src/staking/overview.ts +++ b/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 { return api.query.staking.currentEraPointsEarned(activeEra).pipe( map(({ individual, total }): EraRewardPoints => - createType(api.registry, 'EraRewardPoints', { + api.registry.createType('EraRewardPoints', { total, individual: new Map( 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 combineLatest([ of({ ...indexes, nextElected, validators }), @@ -43,9 +43,10 @@ export function overview (api: ApiInterfaceRx): () => Observable(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 })) diff --git a/packages/api-derive/src/staking/query.ts b/packages/api-derive/src/staking/query.ts index 860fbd2aa5..622cc9056b 100644 --- a/packages/api-derive/src/staking/query.ts +++ b/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 { return memo((accountId: Uint8Array | string): Observable => { - const stashId = createType(api.registry, 'AccountId', accountId); + const stashId = api.registry.createType('AccountId', accountId); return combineLatest([ api.query.session.queuedKeys>>(), @@ -88,6 +88,7 @@ export function query (api: ApiInterfaceRx): (accountId: Uint8Array | string) => ? retrieveCurr(api, stashId) : retrievePrev(api, stashId) ]).pipe( + observeOn(asyncScheduler), switchMap((result): Observable => retrieveController(api, stashId, result) ) diff --git a/packages/api-derive/src/staking/stakerRewards.ts b/packages/api-derive/src/staking/stakerRewards.ts index b964d644a0..78d8bdc291 100644 --- a/packages/api-derive/src/staking/stakerRewards.ts +++ b/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) ) diff --git a/packages/api-derive/src/staking/validators.ts b/packages/api-derive/src/staking/validators.ts index e0c421d2b5..2ddfeff1e0 100644 --- a/packages/api-derive/src/staking/validators.ts +++ b/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 ({ nextElected: nextElected.length ? nextElected diff --git a/packages/api-derive/src/treasury/proposals.ts b/packages/api-derive/src/treasury/proposals.ts index baf706cddc..489534b408 100644 --- a/packages/api-derive/src/treasury/proposals.ts +++ b/packages/api-derive/src/treasury/proposals.ts @@ -6,9 +6,9 @@ import { ApiInterfaceRx } from '@polkadot/api/types'; import { ProposalIndex, TreasuryProposal } from '@polkadot/types/interfaces'; import { DerivedCollectiveProposals, DerivedTreasuryProposal, DerivedTreasuryProposals } from '../types'; -import { Observable, combineLatest, of } from 'rxjs'; -import { map, switchMap } from 'rxjs/operators'; -import { createType, Option } from '@polkadot/types'; +import { Observable, asyncScheduler, combineLatest, of } from 'rxjs'; +import { map, observeOn, switchMap } from 'rxjs/operators'; +import { Option } from '@polkadot/types'; import { memo } from '../util'; @@ -55,7 +55,7 @@ function retrieveProposals (api: ApiInterfaceRx, proposalCount: ProposalIndex, a const isApproval = approvalIds.some((id): boolean => id.eqn(index)); if (!isApproval) { - proposalIds.push(createType(api.registry, 'ProposalIndex', index)); + proposalIds.push(api.registry.createType('ProposalIndex', index)); } } @@ -65,6 +65,7 @@ function retrieveProposals (api: ApiInterfaceRx, proposalCount: ProposalIndex, a api.query.treasury.proposals.multi>(allIds), api.derive.council.proposals() ]).pipe( + observeOn(asyncScheduler), map(([allProposals, councilProposals]: [Option[], DerivedCollectiveProposals]): DerivedTreasuryProposals => parseResult(api, { allIds, allProposals, approvalIds, councilProposals, proposalCount }) ) @@ -81,13 +82,14 @@ export function proposals (api: ApiInterfaceRx): () => Observable retrieveProposals(api, proposalCount, approvalIds) ) ) : of({ approvals: [], - proposalCount: createType(api.registry, 'ProposalIndex'), + proposalCount: api.registry.createType('ProposalIndex'), proposals: [] } as DerivedTreasuryProposals) ); diff --git a/packages/api/src/base/Decorate.ts b/packages/api/src/base/Decorate.ts index b45b1d79ac..431c70ea4f 100644 --- a/packages/api/src/base/Decorate.ts +++ b/packages/api/src/base/Decorate.ts @@ -10,8 +10,8 @@ import { SubmittableExtrinsic } from '../submittable/types'; import { ApiInterfaceRx, ApiOptions, ApiTypes, DecorateMethod, DecoratedRpc, DecoratedRpcSection, QueryableModuleStorage, QueryableStorage, QueryableStorageEntry, QueryableStorageMulti, QueryableStorageMultiArg, SubmittableExtrinsicFunction, SubmittableExtrinsics, SubmittableModuleExtrinsics } from '../types'; import BN from 'bn.js'; -import { BehaviorSubject, Observable, combineLatest, of } from 'rxjs'; -import { map, switchMap, take } from 'rxjs/operators'; +import { BehaviorSubject, Observable, asyncScheduler, combineLatest, of } from 'rxjs'; +import { map, observeOn, switchMap, take } from 'rxjs/operators'; import decorateDerive, { ExactDerive } from '@polkadot/api-derive'; import { memo } from '@polkadot/api-derive/util'; import DecoratedMeta from '@polkadot/metadata/Decorated'; @@ -460,7 +460,10 @@ export default abstract class Decorate extends Events // Since we have a default constructed key, we have Option as the result // Don't keep the subscription here active, retrieve and get out of Dodge // TODO Once we have a multiple key RPC available, use that via fallback - this._rpcCore.state.subscribeStorage[]>(keys).pipe(take(1)) + this._rpcCore.state.subscribeStorage[]>(keys).pipe( + take(1), + observeOn(asyncScheduler) + ) ]) ), map(([keys, values]): [StorageKey, Codec][] =>