import { StreamAction , Feeds , FeedItem } from '../types' ; import * as _ from 'highland' ; private param { Feeds < FeedItem > } feeds return { Feeds < FeedItem > } function createStreams ( feeds ) { const streams = { } ; Object . keys ( feeds ) . forEach ( ( key ) => { const feed = feeds [ key ] ; if ( _ . isStream ( feed ) ) { streams [ key ] = < Highland . Stream < FeedItem > > feed ; } else { streams [ key ] = _ ( feed ) ; } } ) ; return streams ; } private param { Feeds < FeedItem > } feeds return { Highland . Stream < StreamAction > } export function createStreamMerged ( feeds ) { const streams = createStreams ( feeds ) ; return _ ( Object . keys ( streams ) . map ( key => streams [ key ] . map ( ( item ) => ( { type : key , payload : item } ) ) ) ) . merge ( ) ; } private param { Feeds < FeedItem > } feeds return { Highland . Stream < StreamAction > } export function createStreamSorted ( feeds ) { const buffer = new Map ( ) ; const streams = createStreams ( feeds ) ; const pred = ( x1 , x2 ) => { const t1 = x1 . payload && x1 . payload . timestamp ; const t2 = x2 . payload && x2 . payload . timestamp ; if ( typeof t1 === 'undefined' && typeof t2 === 'undefined' ) { return false ; } if ( typeof t2 === 'undefined' ) { return true ; } if ( typeof t1 === 'undefined' ) { return false ; } return t1 < t2 ; } ; const sources = _ ( Object . keys ( streams ) . map ( key => streams [ key ] . map < StreamAction > ( ( item ) => ( { type : key , payload : item } ) ) ) ) as any ; return sources . collect ( ) . flatMap ( ( srcs ) => { const nextValue = ( src , push , next ) => { src . pull ( ( err , x ) => { if ( err ) { push ( err ) ; nextValue ( src , push , next ) ; } else if ( x === _ . nil as any ) { if ( buffer . get ( src ) ) { push ( null , buffer . get ( src ) ) ; } if ( buffer . size <= 1 ) { push ( null , _ . nil ) ; } else { buffer . delete ( src ) ; srcs . splice ( srcs . indexOf ( src ) , 1 ) ; next ( ) ; } } else { if ( buffer . size === srcs . length ) { push ( null , buffer . get ( src ) ) ; } buffer . set ( src , x ) ; next ( ) ; } } ) ; } ; if ( ! srcs . length ) { return _ ( [ ] ) ; } let first = true ; return _ ( ( push , next ) => { if ( first ) { for ( const src of srcs ) { nextValue ( src , push , next ) ; } first = false ; } let srcToPull ; if ( buffer . size === srcs . length ) { for ( const pair of Array . from ( buffer . entries ( ) ) ) { srcToPull = srcToPull === undefined || pred ( pair [ 1 ] , srcToPull [ 1 ] ) ? pair : srcToPull ; } nextValue ( srcToPull [ 0 ] , push , next ) ; } } ) as Highland . Stream < StreamAction > ; } ) ; }