import { logging , settings } from "@cityofzion/neon-core" ; import WebSocket from "isomorphic-ws" ; import { NotificationMessage , CallbackFunction } from "./responses" ; import Subscription from "./subscription" ; const log = logging . default ( "api" ) ; interface ContractSubscriptions { websocket : WebSocket ; callbacks : Map < number , CallbackFunction > ; } export class Notifications { private url : string ; private uniqueIdentifier : number ; public get name ( ) : string { return `Notifications[${ this . url }]` ; } private subscriptions : Map < string | null , ContractSubscriptions > ; param url public constructor ( url ) { this . uniqueIdentifier = 0 ; this . url = settings . networks [ url ] ?. extra ?. notifications ?? url ; this . subscriptions = new Map < string | null , ContractSubscriptions > ( ) ; log . info ( `Created Notifications Provider: ${ this . url }` ) ; } param contract param callback return public subscribe ( contract , callback ) { contract = this . normalizeContract ( contract ) ; const subscriptionIdentifier = this . uniqueIdentifier ++ ; const contractSubscriptions = this . subscriptions . get ( contract ) ?? this . createWebsocketForContract ( contract ) ; contractSubscriptions . callbacks . set ( subscriptionIdentifier , callback ) ; const unsubscribe = ( ) : void => { const callbacksMaps = this . subscriptions . get ( contract ) ?. callbacks ; if ( ! callbacksMaps ?. get ( subscriptionIdentifier ) ) { return ; } if ( callbacksMaps . size > 1 ) { callbacksMaps . delete ( subscriptionIdentifier ) ; } else { this . unsubscribeContract ( contract ) ; } } ; return new Subscription ( contract , unsubscribe ) ; } param contract public unsubscribeContract ( contract ) { contract = this . normalizeContract ( contract ) ; const subscriptionObj = this . subscriptions . get ( contract ) ; if ( ! subscriptionObj || subscriptionObj . websocket . readyState >= WebSocket . CLOSING ) { return Promise . resolve ( ) ; } const closingPromise = new Promise ( ( resolve ) => { subscriptionObj . websocket . onclose = ( ) => { this . subscriptions . delete ( contract ) ; resolve ( ) ; } ; subscriptionObj . websocket . close ( ) ; } ) ; const terminatePromise = new Promise ( ( resolve ) => setTimeout ( resolve , 5000 ) ) . then ( ( ) => { if ( subscriptionObj . websocket . readyState != WebSocket . CLOSED ) { log . warn ( `Websocket for Subscription[${ contract }] did not close in time (was in state ${ subscriptionObj . websocket . readyState }). Force terminating.` ) ; subscriptionObj . websocket . terminate ( ) ; } } ) ; return Promise . race ( [ closingPromise , terminatePromise ] ) ; } /**
     * Unsubscribe all subscriptions (equivalent to calling unsubscribeContract() once for every contract)
     */ public unsubscribeAll ( ) { const closingPromises = [ ] ; for ( const contract of this . subscriptions . keys ( ) ) { closingPromises . push ( this . unsubscribeContract ( contract ) ) ; } return Promise . all ( closingPromises ) . then ( ) ; } private createWebsocketForContract ( contract ) { const ws = new WebSocket ( this . url + ( contract !== null ? "?contract=" + contract : "" ) ) ; ws . onmessage = ( event ) => { const callbackArray = this . subscriptions . get ( contract ) ?. callbacks ?? [ ] ; for ( const cb of callbackArray . values ( ) ) { cb ( JSON . parse ( event . data as string ) as NotificationMessage ) ; } } ; const contractSubscriptions = { websocket : ws , callbacks : new Map < number , CallbackFunction > ( ) , } ; this . subscriptions . set ( contract , contractSubscriptions ) ; return contractSubscriptions ; } private normalizeContract ( contract ) { if ( contract === null ) { return null ; } else if ( contract . slice ( 0 , 2 ) === "0x" ) { return contract ; } else { return "0x" + contract ; } } } export default Notifications ;