import { BufferLike } from '../interfaces' ; import { IterableX } from '../iterable/iterablex' ; import { Readable , ReadableOptions } from 'stream' ; const done = ( _ ) => null as any ; type SourceIterator < TSource > = Iterator < TSource , any , number | ArrayBufferView | undefined | null > ; export class IterableReadable < T > extends Readable { private _pulling = false ; private _objectMode = true ; private _iterator : SourceIterator < T > | undefined ; constructor ( source , options ? ) { super ( options ) ; this . _iterator = source [ Symbol . iterator ] ( ) ; this . _objectMode = ! options || ! ! options . objectMode ; } public _read ( size ) { const it = this . _iterator ; if ( it && ! this . _pulling && ( this . _pulling = true ) ) { this . _pulling = this . _pull ( it , size ) ; } } public _destroy ( err , cb ) { const it = this . _iterator ; this . _iterator = undefined ; const fn = ( it && ( err ? it . throw : it . return ) ) || done ; fn . call ( it , err ) ; if ( cb ) { cb ( null ) ; } } _pull ( it , size ) { let innerSize = size ; const objectMode = this . _objectMode ; let r ; while ( this . readable && ! ( r = it . next ( innerSize ) ) . done ) { if ( innerSize != null ) { if ( objectMode ) { innerSize -= 1 ; } else { innerSize -= Buffer . byteLength ( < BufferLike > r . value || '' ) ; } } if ( ! this . push ( r . value ) || innerSize <= 0 ) { break ; } } if ( ( r && r . done ) || ! this . readable ) { this . push ( null ) ; if ( it . return ) { it . return ( ) ; } } return ! this . readable ; } } export function toNodeStream < TSource > ( source ) ; export function toNodeStream < TSource > ( source , options ) ; export function toNodeStream < TSource extends BufferLike > ( source , options ) ; export function toNodeStream < TSource > ( source , options ? ) { return ! options || options . objectMode === true ? new IterableReadable < TSource > ( source , options ) : new IterableReadable < TSource extends BufferLike ? TSource : any > ( source , options ) ; } ignore export function toNodeStreamProto < TSource > ( this ) ; export function toNodeStreamProto < TSource > ( this , options ) ; export function toNodeStreamProto < TSource extends BufferLike > ( this , options ) ; export function toNodeStreamProto < TSource > ( this , options ? ) { return ! options || options . objectMode === true ? new IterableReadable < TSource > ( this , options ) : new IterableReadable < TSource extends BufferLike ? TSource : any > ( this , options ) ; } IterableX . prototype . toNodeStream = toNodeStreamProto ; declare module '../iterable/iterablex' { interface IterableX < T > { toNodeStream : typeof toNodeStreamProto ; } }