import { Observable , concat } from 'rxjs' ; import { filter , map , scan , take } from 'rxjs/operators' ; import { WebsocketData } from '../websocket-insider' ; export function transformSourceData < T , U > ( params ) { const { source$ , update , transform } = params ; const snapshot$ = source$ . pipe ( filter ( ( orderbookData ) => orderbookData . action === 'partial' ) , take ( 1 ) ) ; const update$ = source$ . pipe ( filter ( ( orderbookData ) => orderbookData . action === 'update' || orderbookData . action === 'insert' || orderbookData . action === 'delete' ) ) ; return update ? concat ( snapshot$ , update$ ) . pipe ( scan ( update ) , map ( transform ) ) : snapshot$ . pipe ( map ( transform ) ) ; }