import WebSocket from 'ws' ; import { Channel , ChannelName } from '../models/Channel' ; import { Order , Side } from '../models/Order' ; import { Orderbook , OrderbookLevel } from '../models/Orderbook' ; import { Ticker } from '../models/Ticker' ; import { Trade } from '../models/Trade' ; import { PriceLevel } from '../models/PriceLevel' ; export interface HydroWatcherOptions { websocketUrl ? : string ; } /**
 * The Hydro API provides a websocket connection that can push updates
 * about the exchange to clients. This watcher lets you create a listener
 * function to handle updates, and a subscription function to choose which
 * channels you wish to get updates on.
 *
 * See https://docs.ddex.io/#websocket
 */ export class HydroWatcher { private static SOCKET_URL : string = 'wss://ws.ddex.io/v3' ; private listener : HydroListener ; private options ? : HydroWatcherOptions ; private socket ? : WebSocket ; private messageQueue : string [ ] = [ ] ; param listener constructor ( listener , options ? ) { this . listener = listener ; this . options = options ; } param channel param marketIds public subscribe ( channel , marketIds ) { this . initIfNeeded ( ) ; this . sendMessage ( this . generateMessage ( 'subscribe' , channel , marketIds ) ) ; } param channel param marketIds public unsubscribe ( channel , marketIds ) { this . initIfNeeded ( ) ; this . sendMessage ( this . generateMessage ( 'unsubscribe' , channel , marketIds ) ) ; } private initIfNeeded ( ) { if ( ! this . socket || this . socket . readyState === WebSocket . CLOSED ) { this . socket = new WebSocket ( this . getWebsocketUrl ( ) ) ; this . socket . on ( 'message' , ( message ) => { this . receiveMessage ( message ) ; } ) ; this . socket . on ( 'close' , ( ) => { this . initIfNeeded ( ) ; } ) ; this . socket . on ( 'open' , ( ) => { while ( this . messageQueue . length > 0 ) { let message = this . messageQueue . shift ( ) ; this . socket && this . socket . send ( message ) ; } } ) ; } } private getWebsocketUrl ( ) { return this . options && this . options . websocketUrl ? this . options . websocketUrl : HydroWatcher . SOCKET_URL ; } private sendMessage ( message ) { if ( this . socket && this . socket . readyState === WebSocket . OPEN ) { this . socket . send ( message ) ; } else { this . messageQueue . push ( message ) ; } } private receiveMessage ( message ) { if ( ! this . listener ) { return ; } const json = JSON . parse ( message ) ; switch ( json . type ) { case 'subscriptions' : return ( this . listener . subscriptionsUpdate && this . listener . subscriptionsUpdate ( json . channels . map ( ( channel ) => new Channel ( channel ) ) ) ) ; case 'ticker' : return this . listener . tickerUpdate && this . listener . tickerUpdate ( new Ticker ( json ) ) ; case 'level2OrderbookSnapshot' : return ( this . listener . orderbookSnapshot && this . listener . orderbookSnapshot ( new Orderbook ( json ) ) ) ; case 'level2OrderbookUpdate' : return ( json . changes && json . changes . forEach ( ( change ) => { this . listener . orderbookUpdate && this . listener . orderbookUpdate ( change . side , new PriceLevel ( change ) ) ; } ) ) ; case 'level3OrderbookSnapshot' : return ( this . listener . fullSnapshot && this . listener . fullSnapshot ( new Orderbook ( json ) , json . sequence ) ) ; case 'receive' : return ( this . listener . orderReceived && this . listener . orderReceived ( new Order ( json ) , json . sequence , new Date ( json . time ) ) ) ; case 'open' : return ( this . listener . orderOpened && this . listener . orderOpened ( new Order ( json ) , json . sequence , new Date ( json . time ) ) ) ; case 'done' : return ( this . listener . orderDone && this . listener . orderDone ( new Order ( json ) , json . sequence , new Date ( json . time ) ) ) ; case 'change' : return ( this . listener . orderChanged && this . listener . orderChanged ( new Order ( json ) , json . sequence , new Date ( json . time ) ) ) ; case 'trade' : return ( this . listener . tradeBegin && this . listener . tradeBegin ( new Trade ( json ) , json . sequence , new Date ( json . time ) ) ) ; case 'trade_success' : return ( this . listener . tradeSuccess && this . listener . tradeSuccess ( new Trade ( json ) , json . sequence , new Date ( json . time ) ) ) ; } } private generateMessage ( type , channel , marketIds ) { return JSON . stringify ( { type : type , channels : [ { name : channel , marketIds : marketIds , } , ] , } ) ; } } export interface HydroListener { /**
     * Received whenever you subscribe or unsubscribe to tell you your
     * current list of channels and market ids you are watching
     */ subscriptionsUpdate ? : ( channels : Channel [ ] ) => void ; /**
     * Received when subscribed to the 'ticker' channel and ticker
     * data is updated
     */ tickerUpdate ? : ( ticker : Ticker ) => void ; /**
     * Received when subscribed to the 'orderbook' channel right
     * after you subscribe
     */ orderbookSnapshot ? : ( orderbook : Orderbook ) => void ; /**
     * Received when subscribed to the 'orderbook' channel and there
     * are changes to the orderbook
     */ orderbookUpdate ? : ( side : Side , priceLevel : PriceLevel ) => void ; /**
     * Received when subscribed to the 'full' channel right after
     * you subscribe
     */ fullSnapshot ? : ( orderbook : Orderbook , sequence : number ) => void ; /**
     * Received when subscribed to the 'full' channel and a new order
     * has been created
     */ orderReceived ? : ( order : Order , sequence : number , time : Date ) => void ; /**
     * Received when subscribed to the 'full' channel and a new order
     * has been created, but not immediately fulfilled
     */ orderOpened ? : ( order : Order , sequence : number , time : Date ) => void ; /**
     * Received when subscribed to the 'full' channel and an order is
     * being taken off the orderbook, either due to being completely
     * fulfilled or because it was cancelled
     */ orderDone ? : ( order : Order , sequence : number , time : Date ) => void ; /**
     * Received when subscribed to the 'full' channel and an order is
     * being updated with new data, usually because it was partially
     * fulfilled
     */ orderChanged ? : ( order : Order , sequence : number , time : Date ) => void ; /**
     * Received when subscribed to the 'full' channel and two orders
     * have been matched, creating a trade
     */ tradeBegin ? : ( trade : Trade , sequence : number , time : Date ) => void ; /**
     * Received when subscribed to the 'full' channel and a trade has
     * been successfully validated on the blockchain
     */ tradeSuccess ? : ( trade : Trade , sequence : number , time : Date ) => void ; }