/**
 * Copyright (c) 2013 Bruno Jouhier <bruno.jouhier@sage.com>
 *
 * Permission is hereby granted, free of charge, to any person
 * obtaining a copy of this software and associated documentation
 * files (the "Software"), to deal in the Software without
 * restriction, including without limitation the rights to use,
 * copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the
 * Software is furnished to do so, subject to the following
 * conditions:
 *
 * The above copyright notice and this permission notice shall be
 * included in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
 * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
 * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
 * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 * OTHER DEALINGS IN THE SOFTWARE.
 */ import { _ } from "streamline-runtime" ; import { convert as predicate } from "./predicate" ; import { Writer } from './writer' ; import * as stopException from './stop-exception' ; import * as nodeStream from "stream" ; import { nextTick } from './util' ; function tryCatch < R > ( _ , that , f ) { try { return f . call ( that , _ ) ; } catch ( ex ) { that . stop ( _ , ex ) ; throw ex ; } } export interface ParallelOptions { count ? : number ; shuffle ? : boolean ; } export interface CompareOptions < T > { compare ? : ( v1 : T , v2 : T ) => number ; } export interface Stoppable { stop : ( _ : _ , arg ? : any ) => void ; } function resolvePredicate < T > ( fn ) { const f = fn ; if ( typeof fn !== 'function' ) return predicate ( fn ) ; else return f ; } export class Reader < T > { parent ? : Stoppable ; read : ( this : Reader < T > , _ : _ ) => T | undefined ; _stop : ( this : Reader < T > , _ : _ , arg ? : any ) => void ; stopped : boolean ; headers : { [ name ] : string ; } ; constructor ( read , stop ? , parent ? ) { if ( typeof read !== 'function' ) throw new Error ( "invalid reader.read: " + ( read && typeof read ) ) ; this . parent = parent ; this . read = read ; this . stopped = false ; if ( stop ) this . _stop = stop ; } forEach ( _ , fn , thisObj ? ) { thisObj = thisObj !== undefined ? thisObj : this ; return tryCatch ( _ , this , ( _ ) => { var i , val ; for ( i = 0 ; ( val = this . read ( _ ) ) !== undefined ; i ++ ) { fn . call ( thisObj , _ , val , i ) ; } return i ; } ) ; } map < U > ( fn , thisObj ? ) { thisObj = thisObj !== undefined ? thisObj : this ; return new Reader ( ( _ ) => { var count = 0 ; var val = this . read ( _ ) ; if ( val === undefined ) return undefined ; return fn . call ( thisObj , _ , val , count ++ ) ; } , undefined , this ) ; } every ( _ , fn , thisObj ? ) { thisObj = thisObj !== undefined ? thisObj : this ; const f = resolvePredicate ( fn ) ; return tryCatch ( _ , this , ( _ ) => { while ( true ) { var val = this . read ( _ ) ; if ( val === undefined ) return true ; if ( ! f . call ( thisObj , _ , val ) ) { this . stop ( _ ) ; return false ; } ; } } ) ; } some ( _ , fn , thisObj ? ) { thisObj = thisObj !== undefined ? thisObj : this ; const f = resolvePredicate ( fn ) ; return tryCatch ( _ , this , ( _ ) => { while ( true ) { var val = this . read ( _ ) ; if ( val === undefined ) return false ; if ( f . call ( thisObj , _ , val ) ) { this . stop ( _ ) ; return true ; } } } ) ; } reduce < U > ( _ , fn , v , thisObj ? ) { thisObj = thisObj !== undefined ? thisObj : this ; return tryCatch ( _ , this , ( _ ) => { while ( true ) { var val = this . read ( _ ) ; if ( val === undefined ) return v ; v = fn . call ( thisObj , _ , v , val ) ; } } ) ; } pipe ( _ , writer ) { tryCatch ( _ , this , ( _ ) => { do { var val = this . read ( _ ) ; try { writer . write ( _ , val ) ; } catch ( ex ) { var arg = stopException . unwrap ( ex ) ; if ( arg && arg !== true ) { this . stop ( _ , arg ) ; throw arg ; } else { this . stop ( _ , arg ) ; break ; } } } while ( val !== undefined ) ; } ) ; return writer ; } tee ( writer ) { const parent = this ; var writeStop ; var readStop ; const stopResult = ( arg ) => { if ( ! arg || arg === true ) return undefined ; else throw arg ; } ; const readDirect = ( _ ) => { var val = parent . read ( _ ) ; if ( ! writeStop ) { try { writer . write ( _ , val ) ; } catch ( ex ) { const arg = stopException . unwrap ( ex ) ; writeStop = [ arg ] ; if ( readStop ) { if ( ! readStop [ 0 ] ) parent . stop ( _ , arg ) ; if ( arg && arg !== true ) throw arg ; else val = undefined ; } else if ( arg ) { readStop = writeStop ; parent . stop ( _ , arg ) ; if ( arg && arg !== true ) throw arg ; else val = undefined ; } } } return val ; } ; return new Reader ( function read ( _ ) { if ( readStop ) return stopResult ( readStop [ 0 ] ) ; return readDirect ( _ ) ; } , function stop ( _ , arg ) { if ( readStop ) return ; readStop = [ arg ] ; if ( arg && ! writeStop ) { writeStop = readStop ; writer . stop ( _ , arg ) ; parent . stop ( _ , arg ) ; } else if ( writeStop && ! writeStop [ 0 ] ) { parent . stop ( _ , arg ) ; } else if ( ! writeStop ) { _ . run ( ( _ ) => { while ( readDirect ( _ ) !== undefined ) ; } ) ; } } , parent ) ; } dup ( ) { const uturn = require ( './devices/uturn' ) . create ( ) ; return [ this . tee ( uturn . writer ) , uturn . reader ] ; } concat ( ... readers ) { const streams = Array . prototype . concat . apply ( [ ] , arguments ) ; var stream = this ; return new Reader ( function read ( _ ) { var val ; while ( stream && ( val = stream . read ( _ ) ) === undefined ) stream = streams . shift ( ) ; return val ; } , function stop ( _ , arg ) { while ( stream ) { stream . stop ( _ , arg ) ; stream = streams . shift ( ) ; } } , this ) ; } toArray ( _ ) { return this . reduce ( _ , ( _ , arr , elt ) => { arr . push ( elt ) ; return arr ; } , [ ] as T [ ] ) ; } readAll ( _ ) { const arr = this . toArray ( _ ) ; if ( typeof arr [ 0 ] === 'string' ) return arr . join ( '' ) ; if ( Buffer . isBuffer ( arr [ 0 ] ) ) { const bufs = arr ; return Buffer . concat ( bufs ) ; } return arr . length > 0 ? arr : undefined ; } transform < U > ( fn , thisObj ? ) { thisObj = thisObj !== undefined ? thisObj : this ; const parent = this ; const uturn = require ( './devices/uturn' ) . create ( ) ; _ . run ( _ => fn . call ( thisObj , _ , parent , uturn . writer ) , err => { _ . run ( _ => parent . stop ( _ ) , e => { uturn . end ( err || e ) ; } ) ; } ) ; return uturn . reader ; } filter ( fn , thisObj ? ) { thisObj = thisObj !== undefined ? thisObj : this ; const f = resolvePredicate ( fn ) ; const parent = this ; var i = 0 , done = false ; return new Reader ( function ( _ ) { while ( ! done ) { var val = parent . read ( _ ) ; done = val === undefined ; if ( done || f . call ( thisObj , _ , val , i ++ ) ) return val ; } return undefined ; } , undefined , parent ) ; } until ( fn , thisObj ? , stopArg ? ) { thisObj = thisObj !== undefined ? thisObj : this ; const f = resolvePredicate ( fn ) ; const parent = this ; var i = 0 ; return new Reader ( function ( _ ) { var val = parent . read ( _ ) ; if ( val === undefined ) return undefined ; if ( ! f . call ( thisObj , _ , val , i ++ ) ) return val ; parent . stop ( _ , stopArg ) ; return undefined ; } , undefined , parent ) ; } while ( fn , thisObj ? , stopArg ? ) { const f = resolvePredicate ( fn ) ; return this . until ( ( _ , val , i ) => ! f . call ( thisObj , _ , val , i ) , thisObj , stopArg ) ; } limit ( n , stopArg ? ) { return this . until ( ( _ , val , i ) => i >= n , this , stopArg ) ; } skip ( n ) { return this . filter ( ( _ , val , i ) => i >= n ) ; } fork ( consumers ) { const parent = this ; const readers = [ ] ; if ( consumers . length === 1 ) { readers . push ( consumers [ 0 ] ( parent ) ) ; } else { var source = parent ; for ( var i = 0 ; i < consumers . length - 1 ; i ++ ) { var dup = source . dup ( ) ; readers . push ( consumers [ i ] ( dup [ 0 ] ) ) ; source = dup [ 1 ] ; } readers . push ( consumers [ consumers . length - 1 ] ( source ) ) ; } return new StreamGroup ( readers ) ; } ; parallel ( options , consumer ) { var opts ; if ( typeof options === "number" ) opts = { count : options , } ; else opts = options ; const parent = this ; const streams = [ ] ; const funnel = _ . funnel ( 1 ) ; var inside = 0 ; var stopArg ; for ( var i = 0 ; i < opts . count ; i ++ ) { ( ( i ) => { streams . push ( consumer ( new Reader ( function read ( _ ) { if ( stopArg ) { if ( stopArg === true ) return undefined ; else throw stopArg ; } return funnel ( _ , ( _ ) => { if ( inside ++ !== 0 ) throw new Error ( "funnel error: " + inside ) ; var val = parent . read ( _ ) ; inside -- ; return val ; } ) ; } , function stop ( _ , arg ) { if ( stopArg ) return ; stopArg = arg ; parent . stop ( _ , arg ) ; } , parent ) ) ) ; } ) ( i ) ; } const group = new StreamGroup ( streams ) ; return opts . shuffle ? group . dequeue ( ) : group . rr ( ) ; } peekable ( ) { const that = this ; return new PeekableReader ( that ) ; } buffer ( max ) { const parent = this ; const buffered = [ ] ; var resume ; var err ; var pending = false ; const fill = ( ) => { if ( pending ) return ; pending = true ; _ . run ( _ => parent . read ( _ ) , ( e , v ) => { pending = false ; if ( e ) err = err || e ; else buffered . push ( v ) ; if ( resume ) { var cb = resume ; resume = undefined ; if ( buffered . length > 0 ) { v = buffered . shift ( ) ; fill ( ) ; cb ( null , v ) ; } else { cb ( err ) ; } } else if ( buffered . length < max ) { if ( ! err && v !== undefined ) fill ( ) ; } } ) ; } ; fill ( ) ; return new Reader ( _ . cast ( function read ( cb ) { if ( buffered . length > 0 ) { var val = buffered . shift ( ) ; fill ( ) ; cb ( null , val ) ; } else { resume = cb ; } } ) , undefined , parent ) ; } join ( streams , thisObj ? ) { const that = this ; const sts = Array . isArray ( streams ) ? streams : [ streams ] ; return new StreamGroup ( [ that ] . concat ( sts ) ) . dequeue ( ) ; } nodify ( ) { const stream = new ( require ( 'stream' ) . Readable ) ( ) ; var pending = false ; const end = ( ) => { stream . push ( null ) ; } ; const more = ( ) => { if ( pending ) return ; var sync = true ; pending = true ; _ . run ( _ => this . read ( _ ) , ( err , result ) => { pending = false ; if ( err ) return stream . emit ( 'error' , err ) ; if ( result === undefined ) { if ( sync ) nextTick ( end ) ; else end ( ) ; return ; } if ( stream . push ( result ) ) { if ( sync ) nextTick ( more ) ; else more ( ) ; } } ) ; sync = false ; } ; stream . _read = ( ) => { more ( ) ; } ; return stream ; } nodeTransform < U > ( duplex ) { return require ( './devices/node' ) . reader ( this . nodify ( ) . pipe ( duplex ) ) ; } compare ( _ , other , options ? ) { const opts = options || { } ; var compare = opts . compare ; if ( ! compare ) compare = ( a , b ) => a === b ? 0 : a < b ? - 1 : + 1 ; var cmp = 0 ; while ( true ) { var data1 = this . read ( _ ) ; var data2 = other . read ( _ ) ; if ( data1 === undefined ) return data2 === undefined ? 0 : - 1 ; if ( data2 === undefined ) return + 1 ; cmp = compare ( data1 , data2 ) ; if ( cmp !== 0 ) return cmp ; } } ; stop ( _ , arg ? ) { if ( this . stopped ) return ; this . stopped = true ; if ( this . _stop ) this . _stop ( _ , arg ) ; else if ( this . parent ) this . parent . stop ( _ , arg ) ; } ; } export class PeekableReader < T > extends Reader < T > { buffered : ( T | undefined ) [ ] ; constructor ( parent ) { super ( ( _ ) => { return this . buffered . length > 0 ? this . buffered . pop ( ) : parent . read ( _ ) ; } , undefined , parent ) ; this . buffered = [ ] ; } unread ( val ) { this . buffered . push ( val ) ; return this ; } peek ( _ ) { var val = this . read ( _ ) ; this . unread ( val ) ; return val ; } } exports . decorate = function ( proto ) { const readerProto = Reader . prototype ; Object . getOwnPropertyNames ( Reader . prototype ) . forEach ( k => { if ( k !== 'constructor' && ! proto [ k ] ) proto [ k ] = readerProto [ k ] ; } ) ; return proto ; } ; export function create < T > ( read , stop ? ) { return new Reader ( read , stop , undefined ) ; } export class StreamGroup < T > implements Stoppable { readers : ( Reader < T > | null ) [ ] ; constructor ( readers ) { this . readers = readers ; } stop ( _ , arg ? ) { this . readers . forEach_ ( _ , ( _ , rd ) => { if ( rd ) rd . stop ( _ , arg ) ; } ) ; } dequeue ( ) { interface Result { i : number ; e : any | undefined ; v : T | undefined ; next : ( ) => void ; } const results = [ ] ; var alive = this . readers . length ; var resume ; this . readers . forEach ( ( stream , i ) => { if ( ! stream ) return ; const next = ( ) => { if ( alive === 0 ) return ; _ . run ( _ => stream . read ( _ ) , ( e , v ) => { if ( ! e && v === undefined ) alive -- ; if ( e || v !== undefined || alive === 0 ) { if ( resume ) { var cb = resume ; resume = undefined ; cb ( e , v ) ; next ( ) ; } else { results . push ( { i : i , e : e , v : v , next : next , } ) ; } } } ) ; } ; next ( ) ; } ) ; return new Reader ( _ . cast ( function read ( cb ) { if ( alive <= 0 ) return cb ( null ) , void 0 ; const res = results . shift ( ) ; if ( res ) { if ( res . next ) res . next ( ) ; return cb ( res . e , res . v ) ; } else { resume = cb ; } } ) , undefined , this ) ; } rr ( ) { interface Entry { i : number ; stream : Reader < T > ; read : ( _ : _ ) => T | undefined ; } const entry = ( stream , i ) => ( { i : i , stream : stream , read : _ . future ( _ => stream . read ( _ ) ) , } ) ; const q = this . readers . map ( entry ) ; return new Reader ( function ( _ ) { var elt ; while ( elt = q . shift ( ) ) { var val = elt . read ( _ ) ; if ( val !== undefined ) { q . push ( entry ( elt . stream , elt . i ) ) ; return val ; } } return undefined ; } , undefined , this ) ; } join ( fn , thisObj ? ) { thisObj = thisObj !== undefined ? thisObj : this ; var last = 0 ; if ( ! fn ) fn = ( ( _ , values ) => { var i = last ; do { i = ( i + 1 ) % values . length ; var v = values [ i ] ; if ( v !== undefined ) { values [ i ] = undefined ; last = i ; return v ; } } while ( i !== last ) ; return undefined ; } ) ; const values = [ ] ; var active = this . readers . length ; var done = false ; var reply ; const joinerCb = ( err , val ) => { if ( err || val === undefined ) { done = true ; return reply && reply ( err , val ) ; } const rep = reply ; reply = undefined ; if ( rep ) rep ( null , val ) ; } ; const callbacks = this . readers . map ( ( reader , i ) => ( ( err , data ) => { if ( active === 0 ) return reply && reply ( ) ; if ( err ) { done = true ; return reply && reply ( err ) ; } values [ i ] = data ; if ( data === undefined ) { this . readers [ i ] = null ; if ( -- active === 0 ) return reply && reply ( ) ; } const vals = values . filter ( ( val ) => val !== undefined ) ; if ( vals . length === active ) { fn . call ( thisObj , joinerCb , values ) ; } } ) ) ; const refill = ( ) => { var count = 0 ; this . readers . forEach ( ( rd , j ) => { if ( rd && values [ j ] === undefined ) { count ++ ; _ . run ( _ => rd . read ( _ ) , callbacks [ j ] ) ; } } ) ; if ( count === 0 ) throw new Error ( "bad joiner: must pick and reset at least one value" ) ; } ; return new Reader ( _ . cast ( function read ( cb ) { if ( done ) { cb ( ) ; return ; } reply = cb ; refill ( ) ; } ) , undefined , this ) ; } }