import { eventsFilterCallback , eventsFilterRx } from '@acala-network/sdk-core' ; import { ApiPromise , ApiRx } from '@polkadot/api' ; import { BlockHash } from '@polkadot/types/interfaces' ; import { BehaviorSubject , Observable , Subject , Subscription , of , from , firstValueFrom } from 'rxjs' ; import { switchMap , filter } from 'rxjs/operators' ; import { NoQueryPath } from './error' ; import { StorageConfigs } from './types' ; export class Storage < T = unknown > { private configs : StorageConfigs ; private subject : Subject < T > ; private subscriber ! : Subscription ; constructor ( configs ) { this . configs = configs ; this . subject = new BehaviorSubject < T > ( undefined as unknown as T ) ; this . doSubscriber ( ) ; } static create < T = unknown > ( configs ) { return new Storage < T > ( configs ) ; } private doSubscriber ( ) { this . subscriber = this . process ( ) . subscribe ( { next : ( data ) => { this . subject . next ( data ) ; } } ) ; } private process ( ) { const { api } = this . configs ; if ( api . type === 'promise' ) { return this . processWithApiPromise ( ) ; } else { return this . processWithApiRx ( ) ; } } private getQueryFN ( api , path , at ? ) { const arr = path . split ( '.' ) ; const start = at ? api . at ( at ) : api ; return arr . reduce ( ( acc , pathItem ) => { return ( acc as any ) [ pathItem ] ; } , start ) ; } private getBlockHash = ( at ? ) : Observable < BlockHash > => { const api = this . configs . api ; if ( ! at ) return of ( '' as unknown as BlockHash ) ; if ( at && typeof at === 'string' ) return of ( at as unknown as BlockHash ) ; if ( api . type === 'promise' ) { return from ( ( async ( ) => { return ( api as ApiPromise ) . rpc . chain . getBlockHash ( at ) ; } ) ( ) ) ; } else { return ( api as ApiRx ) . rpc . chain . getBlockHash ( at ) ; } } ; private processWithApiRx ( ) { const { path , params , at , triggleEvents } = this . configs ; const api = this . configs . api as ApiRx ; const inner = ( atHash ? ) => { const func = this . getQueryFN ( api , path , atHash ) as ( ... params : any [ ] ) => Observable < T > ; if ( func ) { return func ( ... params ) ; } throw new NoQueryPath ( path ) ; } ; if ( triggleEvents ) { return eventsFilterRx ( api , triggleEvents , true ) . pipe ( switchMap ( ( ) => { return this . getBlockHash ( at ) . pipe ( switchMap ( ( hash ) => { return api . rpc . chain . subscribeFinalizedHeads ( ) . pipe ( switchMap ( ( ) => { return inner ( hash . toString ( ) ) ; } ) ) ; } ) ) ; } ) ) ; } return this . getBlockHash ( at ) . pipe ( switchMap ( ( hash ) => inner ( hash . toString ( ) ) ) ) ; } private processWithApiPromise ( ) { const { path , params , at , triggleEvents } = this . configs ; const api = this . configs . api as ApiPromise ; return new Observable ( ( subscriber ) => { ( async ( ) => { const atHash = at ? await api . rpc . chain . getBlockHash ( at ) : '' ; const func = this . getQueryFN ( api , path , atHash . toString ( ) ) as ( ... params : any [ ] ) => void ; params . push ( ( result ) => { subscriber . next ( result ) ; } ) ; if ( triggleEvents ) { eventsFilterCallback ( api , triggleEvents , true , ( ) => { api . rpc . chain . subscribeFinalizedHeads ( ( ) => { func ( ... params ) ; } ) ; } ) ; } else { func ( ... params ) ; } } ) ( ) ; } ) ; } public unsub ( ) { this . subscriber . unsubscribe ( ) ; } get observable ( ) : Observable < T > { return this . subject . asObservable ( ) . pipe ( filter ( ( i ) => ! ! i ) ) ; } public async query ( ) { return firstValueFrom ( this . observable ) ; } }