import { debug } from './debug' ; import { getFilters , normalizeMessages } from './handy' ; import { MapperFactory } from './mappers' ; import { createRealTimeFeed } from './realtimefeeds' ; import { Disconnect , Exchange , Filter , FilterForExchange } from './types' ; async function * _stream < T extends Exchange , U extends boolean = false > ( { exchange , filters , timeoutIntervalMS = 10000 , withDisconnects = undefined , onError = undefined } ) { validateStreamOptions ( filters ) ; const realTimeFeed = createRealTimeFeed ( exchange , filters , timeoutIntervalMS , onError ) ; for await ( const message of realTimeFeed ) { if ( message . __disconnect__ === true ) { if ( withDisconnects ) { yield undefined as any ; } } else { yield { localTimestamp : new Date ( ) , message } as any ; } } } export function stream < T extends Exchange , U extends boolean = false > ( { exchange , filters , timeoutIntervalMS = 10000 , withDisconnects = undefined , onError = undefined } ) { let _iterator = _stream ( { exchange , filters , timeoutIntervalMS , withDisconnects , onError } ) ; ( _iterator as any ) . __realtime__ = true ; return _iterator ; } async function * _streamNormalized < T extends Exchange , U extends MapperFactory < T , any > [ ] , Z extends boolean = false > ( { exchange , symbols , timeoutIntervalMS = 10000 , withDisconnectMessages = undefined , onError = undefined } , ... normalizers ) { while ( true ) { try { const createMappers = ( localTimestamp ) => normalizers . map ( ( m ) => m ( exchange , localTimestamp ) ) ; const mappers = createMappers ( new Date ( ) ) ; const filters = getFilters ( mappers , symbols ) ; const messages = _stream ( { exchange , withDisconnects : true , timeoutIntervalMS , filters , onError } ) ; const upperCaseSymbols = symbols !== undefined ? symbols . map ( ( s ) => s . toUpperCase ( ) ) : undefined ; const filter = ( symbol ) => { return upperCaseSymbols === undefined || upperCaseSymbols . length === 0 || upperCaseSymbols . includes ( symbol ) ; } ; const normalizedMessages = normalizeMessages ( exchange , messages , mappers , createMappers , withDisconnectMessages , filter , new Date ( ) ) ; for await ( const message of normalizedMessages ) { yield message ; } } catch ( error ) { if ( onError !== undefined ) { onError ( error ) ; } debug ( '%s normalize messages error: %o, retrying with new connection...' , exchange , error ) ; if ( withDisconnectMessages ) { const disconnect = { type : 'disconnect' , exchange , localTimestamp : new Date ( ) } ; yield disconnect as any ; } } } } function validateStreamOptions ( filters ) { if ( ! filters ) { throw new Error ( `Invalid "filters" argument. Please provide filters array` ) ; } for ( let i = 0 ; i < filters . length ; i ++ ) { const filter = filters [ i ] ; if ( filter . symbols && Array . isArray ( filter . symbols ) === false ) { throw new Error ( `Invalid "filters[].symbols" argument: ${ filter . symbols }. Please provide array of symbol strings` ) ; } } } export type StreamOptions < T extends Exchange , U extends boolean = false > = { exchange : T ; filters : FilterForExchange [ T ] [ ] ; timeoutIntervalMS ? : number ; withDisconnects ? : U ; onError ? : ( error : Error ) => void ; } ; export type StreamNormalizedOptions < T extends Exchange , U extends boolean = false > = { exchange : T ; symbols ? : string [ ] ; timeoutIntervalMS ? : number ; withDisconnectMessages ? : U ; onError ? : ( error : Error ) => void ; } ; export function streamNormalized < T extends Exchange , U extends MapperFactory < T , any > [ ] , Z extends boolean = false > ( { exchange , symbols , timeoutIntervalMS = 10000 , withDisconnectMessages = undefined , onError = undefined } , ... normalizers ) { let _iterator = _streamNormalized ( { exchange , symbols , timeoutIntervalMS , withDisconnectMessages , onError } , ... normalizers ) ; ( _iterator as any ) . __realtime__ = true ; return _iterator ; }