import { Observable , ReplaySubject } from 'rxjs' ; import { v4 as uuid } from 'uuid' ; import { Config } from '../../../types' ; import { PrivateEndPoints , PublicEndPoints } from '../../types' ; import { WebsocketData , WebsocketRequest , WebsocketResponse } from './types' ; import { WebsocketInsiderBase } from './websocket-insider-base' ; interface StreamData { uuid : string ; name : string ; stream : ReplaySubject < WebsocketData > ; } export class WebsocketInsider extends WebsocketInsiderBase < WebsocketRequest , WebsocketResponse | WebsocketData > { private streamTable : StreamData [ ] = [ ] ; constructor ( config ) { super ( config ) ; } param message handleMessage ( message ) { const wsData = message as WebsocketData ; if ( wsData . table ) { const symbol = wsData . data && wsData . data . length ? wsData . data [ 0 ] . symbol : '' ; const key = `${ wsData . table }:${ symbol }` ; const streamData = this . streamTable . find ( ( o ) => o . name === key ) ; const allStreamData = this . streamTable . find ( ( o ) => o . name === wsData . table ) ; if ( streamData ) { streamData . stream . next ( wsData ) ; } else if ( allStreamData ) { allStreamData . stream . next ( wsData ) ; } } } subscribe < T > ( channel ) { this . send ( { op : 'subscribe' , args : channel } ) ; return this . fetchOrNewStream ( channel ) ; } param channel unsubscribe ( channel ) { this . send ( { op : 'unsubscribe' , args : channel } ) ; if ( channel instanceof Array ) { for ( const cn of channel ) { this . delStream ( cn ) ; } } else { this . delStream ( channel ) ; } } onDestroy ( ) { const uniqueStreamData = [ ] ; for ( const steamData of this . streamTable ) { if ( uniqueStreamData . length === 0 ) { uniqueStreamData . push ( steamData ) ; } else { const sameData = uniqueStreamData . find ( ( o ) => o . uuid === steamData . uuid ) ; if ( ! sameData ) { uniqueStreamData . push ( steamData ) ; } } } for ( const steamData of uniqueStreamData ) { steamData . stream . complete ( ) ; this . send ( { op : 'unsubscribe' , args : name } ) ; } this . streamTable = [ ] ; } private fetchOrNewStream < T > ( channel ) { let stream ; if ( channel instanceof Array ) { for ( const cn of channel ) { const result = this . streamTable . find ( ( o ) => o . name === cn ) ; if ( result ) { return result . stream ; } } stream = new ReplaySubject < WebsocketData < T > > ( 1 ) ; const _uuid = uuid ( ) ; for ( const cn of channel ) { this . streamTable . push ( { uuid : _uuid , name : cn , stream , } ) ; } return stream ; } const res = this . streamTable . find ( ( o ) => o . name === channel ) ; if ( ! res ) { stream = new ReplaySubject < WebsocketData < T > > ( 1 ) ; this . streamTable . push ( { uuid : uuid ( ) , name : channel , stream , } ) ; } else { stream = res . stream ; } return stream ; } private delStream ( streamName ) { const streamData = this . streamTable . find ( ( data , index ) => { if ( data . name === streamName ) { this . streamTable . splice ( index , 1 ) ; return true ; } return false ; } ) ; if ( streamData ) { const otherSteam = this . streamTable . find ( ( o ) => o . uuid === streamData . uuid ) ; if ( ! otherSteam ) { streamData . stream . complete ( ) ; } } else { const ep = [ ... Object . values ( PrivateEndPoints ) , ... Object . values ( PublicEndPoints ) ] . find ( ( endpoint ) => endpoint === streamName ) ; if ( ep ) { for ( const [ index , data ] of this . streamTable . entries ( ) ) { if ( data . name . includes ( ep ) ) { data . stream . complete ( ) ; this . streamTable . splice ( index , 1 ) ; } } } } } }