import { SocketIORxjs } from '@dripjs/common' ; import { Bar , BarRequest , SupportedExchange } from '@dripjs/types' ; import { Observable } from 'rxjs' ; import { filter } from 'rxjs/operators' ; import { IntelChannel , IntelRealtimeResponse } from '../service' ; export interface IntelClientOptions { ip : string ; port : number ; username : string ; password : string ; } export class IntelClient { private readonly ioRxjs : SocketIORxjs ; constructor ( private readonly options ) { this . ioRxjs = new SocketIORxjs ( `http://${ this . options . ip }:${ this . options . port }` , { transportOptions : { polling : { extraHeaders : { username : this . options . username , password : this . options . password , } , } , } , } ) ; } disconnect ( ) { this . ioRxjs . close ( ) ; } async getSymbols ( exchange ) { return new Promise ( ( resolve ) => { this . ioRxjs . emit ( 'symbols' , exchange , resolve ) ; } ) ; } async getBars ( exchange , barRequest ) { return new Promise ( ( resolve ) => { this . ioRxjs . emit ( 'bars' , { exchange , ... barRequest } , resolve ) ; } ) ; } ticker$ ( exchange , symbol ) { const channel = IntelChannel . Ticker ; return this . subscribe ( channel , exchange , symbol ) ; } stopTicker ( exchange , symbol ) { const channel = IntelChannel . Ticker ; this . unsubscribe ( channel , exchange , symbol ) ; } depth$ ( exchange , symbol ) { const channel = IntelChannel . Depth ; return this . subscribe ( channel , exchange , symbol ) ; } stopDepth ( exchange , symbol ) { const channel = IntelChannel . Depth ; this . unsubscribe ( channel , exchange , symbol ) ; } transaction$ ( exchange , symbol ) { const channel = IntelChannel . Transaction ; return this . subscribe ( channel , exchange , symbol ) ; } stopTransaction ( exchange , symbol ) { const channel = IntelChannel . Transaction ; this . unsubscribe ( channel , exchange , symbol ) ; } private subscribe ( channel , exchange , symbol ) { this . ioRxjs . socket . on ( channel , ( res ) => this . ioRxjs . next ( res ) ) ; this . ioRxjs . emit ( 'subscribe' , { exchange , symbol , channel } ) ; return this . ioRxjs . message$ . pipe ( filter ( ( res ) => res . channel === channel ) ) ; } private unsubscribe ( channel , exchange , symbol ) { this . ioRxjs . emit ( 'unsubscribe' , { exchange , symbol , channel } ) ; } }