import { injectable , inject } from 'inversify' ; import symbols from './symbols' ; import { Quote , ConfigStore , BrokerPosition , BrokerMap , SpreadAnalysisResult , ConfigRoot , LimitCheckResult } from './types' ; import QuoteAggregator from './QuoteAggregator' ; import { getLogger } from '@bitr/logger' ; import * as WebSocket from 'ws' ; import { wssPort } from './constants' ; import * as _ from 'lodash' ; import PositionService from './PositionService' ; import OppotunitySearcher from './OpportunitySearcher' ; import OrderService from './OrderService' ; import OrderImpl from './OrderImpl' ; import * as express from 'express' ; import * as http from 'http' ; import { autobind } from 'core-decorators' ; const opn = require ( 'opn' ) ; @ injectable ( ) @ autobind export default class WebGateway { private readonly eventMapper : [ any , string , any ] [ ] ; private server : http . Server ; private app : express . Express ; private wss : WebSocket . Server ; private readonly log = getLogger ( this . constructor . name ) ; private readonly clients : WebSocket [ ] = [ ] ; private readonly staticPath : string = `${ process . cwd ( ) }/webui/dist` ; constructor ( private readonly quoteAggregator , @ inject ( symbols . ConfigStore ) private readonly configStore , private readonly positionService , private readonly opportunitySearcher , private readonly orderService ) { this . eventMapper = [ [ this . quoteAggregator , 'quoteUpdated' , this . quoteUpdated ] , [ this . positionService , 'positionUpdated' , this . positionUpdated ] , [ this . opportunitySearcher , 'spreadAnalysisDone' , this . spreadAnalysisDone ] , [ this . opportunitySearcher , 'limitCheckDone' , this . limitCheckDone ] , [ this . opportunitySearcher , 'activePairRefresh' , this . activePairRefresh ] , [ this . orderService , 'orderCreated' , this . orderCreated ] , [ this . orderService , 'orderUpdated' , this . orderUpdated ] , [ this . orderService , 'orderFinalized' , this . orderFinalized ] , [ this . configStore , 'configUpdated' , this . configUpdated ] ] ; } async start ( ) { const { webGateway } = this . configStore . config ; if ( ! webGateway || ! webGateway . enabled ) { return ; } const host = _ . defaultTo ( webGateway . host , 'localhost' ) ; this . log . debug ( `Starting ${ this . constructor . name }...` ) ; for ( const e of this . eventMapper ) { e [ 0 ] . on ( e [ 1 ] , e [ 2 ] ) ; } this . app = express ( ) ; this . app . use ( express . static ( this . staticPath ) ) ; this . app . get ( '*' , ( req , res ) => { res . sendFile ( `${ this . staticPath }/index.html` ) ; } ) ; this . server = this . app . listen ( wssPort , host , ( ) => { this . log . debug ( `Express started listening on ${ wssPort }.` ) ; } ) ; this . wss = new WebSocket . Server ( { server : this . server } ) ; this . wss . on ( 'connection' , ws => { ws . on ( 'error' , err => { this . log . debug ( err . message ) ; } ) ; this . clients . push ( ws ) ; } ) ; if ( webGateway . openBrowser ) { opn ( `http://${ host }:${ wssPort }` ) ; } this . log . debug ( `Started ${ this . constructor . name }.` ) ; } async stop ( ) { const { webGateway } = this . configStore . config ; if ( ! webGateway || ! webGateway . enabled ) { return ; } this . log . debug ( `Stopping ${ this . constructor . name }...` ) ; this . wss . close ( ) ; this . server . close ( ) ; for ( const e of this . eventMapper ) { e [ 0 ] . removeListener ( e [ 1 ] , e [ 2 ] ) ; } this . log . debug ( `Stopped ${ this . constructor . name }.` ) ; } private async quoteUpdated ( quotes ) { this . broadcast ( 'quoteUpdated' , quotes ) ; } private positionUpdated ( positions ) { this . broadcast ( 'positionUpdated' , positions ) ; } private spreadAnalysisDone ( result ) { this . broadcast ( 'spreadAnalysisDone' , result ) ; } private limitCheckDone ( limitCheckResult ) { this . broadcast ( 'limitCheckDone' , limitCheckResult ) ; } private async activePairRefresh ( pairsWithAnalysis ) { this . broadcast ( 'activePairRefresh' , pairsWithAnalysis ) ; } private orderCreated ( order ) { this . broadcast ( 'orderCreated' , order ) ; } private orderUpdated ( order ) { this . broadcast ( 'orderUpdated' , order ) ; } private orderFinalized ( order ) { this . broadcast ( 'orderFinalized' , order ) ; } private configUpdated ( config ) { this . broadcast ( 'configUpdated' , this . sanitize ( config ) ) ; } private sanitize ( config ) { const copy = _ . cloneDeep ( config ) ; for ( const brokerConfig of copy . brokers ) { delete brokerConfig . key ; delete brokerConfig . secret ; } delete copy . logging ; return copy ; } private broadcast ( type , body ) { for ( const client of this . clients ) { if ( client . readyState === WebSocket . OPEN ) { client . send ( JSON . stringify ( { type , body } ) , err => { if ( err ) { this . log . debug ( err . message ) ; _ . pull ( this . clients , client ) ; } } ) ; } } } }