import { BehaviorSubject , Observable , Subject } from 'rxjs/Rx' ; import { Client } from 'rpc-websockets' ; import { RadixAID , RadixAtom , RadixSerializer , RadixUniverseConfig } from '../atommodel' ; import { logger } from '../common/RadixLogger' ; import events from 'events' ; import { RadixAtomNodeStatus , RadixAtomNodeStatusUpdate , RadixNode , radixUniverse } from '../..' ; import { RadixUniverseType , universeTypeToString } from '../atommodel/universe/RadixUniverseConfig' ; import { share } from 'rxjs/operators' ; interface Notification { subscriberId : number ; } export interface AtomReceivedNotification extends Notification { atomEvents : any [ ] ; isHead : boolean ; } interface AtomStatusNotification extends Notification { status : string ; data ? : { pointerToIssue ? : string ; message : string ; } ; } interface AtomSubmissionStateUpdateNotification extends Notification { value : string ; data ? : { pointerToIssue ? : string ; message : string ; } ; } export declare interface RadixNodeConnection { on ( event , listener ) : this ; } export class RadixNodeConnection extends events . EventEmitter { private pingInterval ; private _socket : Client ; private _subscriptions : { [ subscriberId ] : Subject < AtomReceivedNotification > ; } = { } ; private _atomUpdateSubjects : { [ subscriberId ] : BehaviorSubject < RadixAtomNodeStatusUpdate > ; } = { } ; private _addressSubscriptions : { [ address ] : string ; } = { } ; private lastSubscriberId = 1 ; public address : string ; constructor ( readonly node ) { super ( ) ; this . node = node ; } private getSubscriberId ( ) { this . lastSubscriberId ++ ; return this . lastSubscriberId + '' ; } returns public isReady ( ) { return this . _socket && this . _socket . ready ; } private ping = ( ) => { if ( this . isReady ( ) ) { this . _socket . call ( 'Ping' , { id : 0 } ) . then ( ( response ) => { logger . debug ( `Ping` , response ) ; } ) . catch ( ( error ) => { logger . warn ( `Error sending ping` , error ) ; } ) ; } } ; returns public async openConnection ( ) { return new Promise < void > ( ( resolve , reject ) => { this . address = this . node . wsAddress ; logger . info ( `Connecting to ${ this . address }` ) ; this . _socket = new Client ( this . address , { reconnect : false } ) ; this . _socket . on ( 'close' , this . _onClosed ) ; this . _socket . on ( 'error' , error => { logger . error ( error ) ; reject ( error ) ; } ) ; setTimeout ( ( ) => { if ( ! this . _socket . ready ) { logger . debug ( 'Socket timeout' ) ; this . _socket . close ( ) ; this . emit ( 'closed' ) ; reject ( 'Timeout' ) ; } } , 5000 ) ; this . _socket . on ( 'open' , ( ) => { logger . info ( `Connected to ${ this . address }` ) ; this . pingInterval = setInterval ( this . ping , 10000 ) ; this . emit ( 'open' ) ; this . _socket . call ( 'Universe.getUniverse' , { } ) . then ( ( response ) => { const nodeUniverseConfig = new RadixUniverseConfig ( response ) ; nodeUniverseConfig . initialize ( ) ; const localUniverseConfig = radixUniverse . universeConfig ; const nodeHid = nodeUniverseConfig . getHidString ( ) ; const localHid = localUniverseConfig . getHidString ( ) ; if ( nodeHid !== localHid ) { if ( localUniverseConfig . type === RadixUniverseType . DEVELOPMENT && nodeUniverseConfig . type === RadixUniverseType . DEVELOPMENT ) { logger . error ( `
                                Please change your tests to be using
                                 'bootstrapTrustedNode(RadixUniverse.LOCAL)'
                                  instead of bootstrap(RadixUniverse.LOCAL).
                                  ` ) ; this . close ( ) ; return ; } if ( localUniverseConfig . type !== nodeUniverseConfig . type ) { const localUniverseTypeName = universeTypeToString ( localUniverseConfig . type ) ; const nodeUniverseTypeName = universeTypeToString ( nodeUniverseConfig . type ) ; logger . error ( `ERROR mismatch between RadixUniverse types of local and node@${ this . address }.
                                Check your local universe config.` , { localUniverseTypeName , nodeUniverseTypeName , } ) ; this . close ( ) ; return ; } logger . error ( `ERROR: Universe configuration mismatch while connecting to node ${ this . address }.
                                 Check your local universe config.` , { localHid , nodeHid , } ) ; this . close ( ) ; } } ) ; this . _socket . on ( 'Atoms.subscribeUpdate' , this . _onAtomReceivedNotification ) ; this . _socket . on ( 'AtomSubmissionState.onNext' , this . _onAtomSubmissionStateUpdate ) ; this . _socket . on ( 'Atoms.nextStatusEvent' , this . _onAtomStatusNotification ) ; resolve ( ) ; } ) ; } ) ; } param address returns public subscribe ( address ) { const subscriberId = this . getSubscriberId ( ) ; this . _addressSubscriptions [ address ] = subscriberId ; this . _subscriptions [ subscriberId ] = new Subject < AtomReceivedNotification > ( ) ; this . _socket . call ( 'Atoms.subscribe' , { subscriberId , query : { address , } , debug : true , } ) . then ( ( response ) => { logger . info ( `Subscribed for address ${ address }` , response ) ; } ) . catch ( ( error ) => { logger . error ( `Error subscribing for address ${ address }` , error ) ; this . _subscriptions [ subscriberId ] . error ( error ) ; } ) ; return this . _subscriptions [ subscriberId ] . pipe ( share ( ) ) ; } param address returns public unsubscribe ( address ) { const subscriberId = this . _addressSubscriptions [ address ] ; return new Promise < any > ( ( resolve , reject ) => { this . _socket . call ( 'Atoms.cancel' , { subscriberId , } ) . then ( ( response ) => { logger . info ( `Unsubscribed for address ${ address }` ) ; this . _subscriptions [ this . _addressSubscriptions [ address ] ] . complete ( ) ; delete this . _addressSubscriptions [ address ] ; resolve ( response ) ; } ) . catch ( ( error ) => { reject ( error ) ; } ) ; } ) ; } returns public unsubscribeAll ( ) { return new Promise < any > ( ( resolve , reject ) => { const unsubscriptions = new Array < Promise < any > > ( ) ; for ( const address in this . _addressSubscriptions ) { unsubscriptions . push ( this . unsubscribe ( address ) ) ; } Promise . all ( unsubscriptions ) . then ( ( values ) => { resolve ( values ) ; } ) . catch ( ( error ) => { reject ( error ) ; } ) ; } ) ; } param atom returns public submitAtom ( atom ) { const subscriberId = this . getSubscriberId ( ) ; const atomStateSubject = new BehaviorSubject ( { status : RadixAtomNodeStatus . PENDING } ) ; this . _atomUpdateSubjects [ subscriberId ] = atomStateSubject ; const timeout = setTimeout ( ( ) => { this . _socket . close ( ) ; atomStateSubject . error ( 'Socket timeout' ) ; } , 5000 ) ; this . _socket . call ( 'Atoms.getAtomStatusNotifications' , { subscriberId , aid : atom . getAidString ( ) , } ) . then ( _ => { const atomJSON = RadixSerializer . toJSON ( atom ) ; return this . _socket . call ( 'Atoms.submitAtom' , atomJSON ) ; } ) . then ( ( response ) => { const aidFromNode = RadixAID . fromAIDString ( response . aid ) ; if ( aidFromNode . toString ( ) !== atom . getAid ( ) . toString ( ) ) { throw new Error ( `AID from Node "${ aidFromNode . toString ( ) }" does not match the local aid "${ atom . getAid ( ) . toString ( ) }".
This is a radixdlt-js issue, please report this at https://github.com/radixdlt/radixdlt-js/issues . 
The atom may or may not have been accepted by the node.
                    ` ) ; } clearTimeout ( timeout ) ; } ) . catch ( ( error ) => { clearTimeout ( timeout ) ; atomStateSubject . error ( error ) ; } ) ; return atomStateSubject . pipe ( share ( ) ) ; } public close = async ( ) => { await this . unsubscribeAll ( ) ; this . _socket . close ( 1001 ) ; clearInterval ( this . pingInterval ) ; } ; private _onClosed = ( ) => { logger . info ( 'Socket closed' ) ; clearInterval ( this . pingInterval ) ; this . emit ( 'closed' ) ; } ; private _onAtomStatusNotification = ( notification ) => { logger . info ( 'Atom Status notification' , notification ) ; const subscriberId = notification . subscriberId ; const subject = this . _atomUpdateSubjects [ subscriberId ] ; const value = notification . status ; if ( ! subject . closed && ! subject . isStopped ) { subject . next ( { status : RadixAtomNodeStatus [ value ] , data : notification . data } ) ; } else { logger . error ( `☢️ subject closed or stopped` ) ; } } ; private _onAtomSubmissionStateUpdate = ( notification ) => { logger . info ( 'Atom Submission state update' , notification ) ; } ; private _onAtomReceivedNotification = ( notification ) => { logger . debug ( 'Atoms notification' , notification ) ; const subscription = this . _subscriptions [ notification . subscriberId ] ; if ( ! subscription . closed ) { subscription . next ( notification ) ; } else { logger . error ( `☢️ subscription closed` ) ; } } ; } export default RadixNodeConnection ;