import { Subject , Observable , Observer , Subscription } from 'rxjs' ; import { TSMap } from 'typescript-map' ; import { filter } from 'rxjs/operators' ; import RadixAccountSystem from './RadixAccountSystem' ; import RadixApplicationDataUpdate from './RadixApplicationDataUpdate' ; import RadixApplicationData from './RadixApplicationData' ; import { RadixAtom , RadixAtomUpdate , RadixAddress , RadixSpin } from '../atommodel' ; import { logger , RadixAtomObservation , RadixAtomStatusIsInsert } from '../..' ; export default class RadixDataAccountSystem implements RadixAccountSystem { public name = 'DATA' ; public applicationDataSubject : Subject < RadixApplicationDataUpdate > = new Subject ( ) ; public applicationData : TSMap < string , TSMap < string , RadixApplicationData > > = new TSMap ( ) ; private subs = new Subscription ( ) ; constructor ( readonly address ) { } public async processAtomUpdate ( atomUpdate ) { if ( ! ( 'decryptedData' in atomUpdate . processedData ) ) { return ; } if ( RadixAtomStatusIsInsert [ atomUpdate . status . status ] ) { this . processStoreAtom ( atomUpdate ) ; } else { this . processDeleteAtom ( atomUpdate ) ; } } public unsubscribeSubscribers ( ) { this . subs . unsubscribe ( ) ; } private processStoreAtom ( atomUpdate ) { const atom = atomUpdate . atom ; const aid = atom . getAidString ( ) ; const signatures = atom . signatures ; const applicationId = atomUpdate . processedData . decryptedData . application ; if ( this . applicationData . has ( applicationId ) && this . applicationData . get ( applicationId ) . has ( aid ) ) { return ; } const applicationData = { aid , payload : atomUpdate . processedData . decryptedData , signatures , } ; const applicationDataUpdate = { action : 'STORE' , aid , applicationId , data : applicationData , signatures , } ; if ( ! this . applicationData . has ( applicationId ) ) { this . applicationData . set ( applicationId , new TSMap ( ) ) ; } this . applicationData . get ( applicationId ) . set ( aid , applicationData ) ; if ( ! this . applicationDataSubject . closed && ! this . applicationDataSubject . isStopped ) { this . applicationDataSubject . next ( applicationDataUpdate ) ; } else { logger . error ( `☢️ applicationDataSubject closed or stopped` ) ; } } private processDeleteAtom ( atomUpdate ) { const atom = atomUpdate . atom ; const aid = atom . getAidString ( ) ; const signatures = atom . signatures ; const applicationId = atomUpdate . processedData . decryptedData . application ; if ( ! this . applicationData . has ( applicationId ) || ! this . applicationData . get ( applicationId ) . has ( aid ) ) { return ; } const applicationData = this . applicationData . get ( applicationId ) . get ( aid ) ; const applicationDataUpdate = { action : 'DELETE' , aid , applicationId , data : applicationData , signatures , } ; this . applicationData . get ( applicationId ) . delete ( aid ) ; if ( ! this . applicationDataSubject . closed && ! this . applicationDataSubject . isStopped ) { this . applicationDataSubject . next ( applicationDataUpdate ) ; } else { logger . error ( `☢️ applicationDataSubject closed or stopped` ) ; } } param applicationId param addresses returns public getApplicationData ( applicationId , addresses ? ) { const signatureIds = ! addresses ? undefined : addresses . map ( a => RadixAddress . fromAddress ( a ) . getUID ( ) . toString ( ) ) ; return Observable . create ( ( observer ) => { if ( this . applicationData . has ( applicationId ) ) { for ( const applicationData of this . applicationData . get ( applicationId ) . values ( ) ) { if ( ! signatureIds || signatureIds . length === 0 || signatureIds . some ( s => Object . keys ( applicationData . signatures ) . includes ( s ) ) ) { const applicationDataUpdate = { action : 'STORE' , aid : applicationData . aid , applicationId , data : applicationData , signatures : applicationData . signatures , } ; if ( ! observer . closed ) { observer . next ( applicationDataUpdate ) ; } else { logger . error ( `☢️ observer closed or stopped` ) ; } } } } const subscription = this . applicationDataSubject . pipe ( filter ( update => { return update . applicationId === applicationId && ( ! signatureIds || signatureIds . length === 0 || signatureIds . some ( s => Object . keys ( update . signatures ) . includes ( s ) ) ) ; } ) ) . subscribe ( observer ) ; this . subs . add ( subscription ) ; } ) ; } }