/**
 * Promise-based object stream with seamless support for back-pressure and error
 * handling, written in Typescript.
 *
 * Copyright (C) 2015 Martin Poelstra
 * License: MIT
 */ import * as fs from "fs" ; import * as NodeStream from "stream" ; import { Readable , Stream } from "./Stream" ; import { defer , swallowErrors , VoidDeferred } from "./util" ; see export class NodeReadable < T > extends NodeStream . Readable { private _resumer ? : ( value ? : void | PromiseLike < void > ) => void ; see param tsReadable param options constructor ( tsReadable , options ? ) { super ( options ) ; swallowErrors ( tsReadable . forEach ( ( chunk ) : void | Promise < void > => { if ( this . push ( chunk ) ) { return ; } const d = defer ( ) ; this . _resumer = d . resolve ; return d . promise ; } , ( err ? ) => { if ( err ) { this . emit ( "error" , err ) ; } this . push ( null ) ; } , ( abortError ) => { if ( this . _resumer ) { this . _resumer ( Promise . reject ( abortError ) ) ; this . _resumer = undefined ; } } ) ) ; } public _read ( size ) { if ( this . _resumer ) { this . _resumer ( ) ; this . _resumer = undefined ; } } } /**
 * Convenience wrapper around Node's file stream.
 *
 * Usage example:
 * let source = Stream.from(["abc", "def"]);
 * source.pipe(new FileSink("test.txt"));
 *
 * To wait for the stream's result, use e.g.
 * let sink = source.pipe(new FileSink("test.txt"));
 * sink.result().then(() => console.log("ok"), (err) => console.log("error", err));
 */ export class FileSink extends Stream < string > { see param path param options constructor ( path , options ? ) { super ( ) ; pipeToNodeStream ( this , fs . createWriteStream ( path , options ) ) ; } } see param tsReadable param nodeWritable param emitError return export function pipeToNodeStream < T > ( tsReadable , nodeWritable , emitError = false ) { const endDeferred = defer ( ) ; let blockedDeferred ; function handleNodeStreamError ( error ) { emitError = false ; endDeferred . reject ( error ) ; tsReadable . abort ( error ) ; if ( blockedDeferred ) { blockedDeferred . reject ( error ) ; nodeWritable . removeListener ( "drain" , blockedDeferred . resolve ) ; } } function handleTsStreamError ( error ) { if ( ! emitError ) { return ; } emitError = false ; nodeWritable . removeListener ( "error" , handleNodeStreamError ) ; nodeWritable . emit ( "error" , error ) ; } nodeWritable . once ( "error" , handleNodeStreamError ) ; nodeWritable . once ( "finish" , ( ) => { nodeWritable . removeListener ( "error" , handleNodeStreamError ) ; endDeferred . resolve ( ) ; } ) ; return tsReadable . forEach ( ( chunk ) : void | Promise < void > => { blockedDeferred = undefined ; const canAcceptMore = nodeWritable . write ( chunk ) ; if ( ! canAcceptMore ) { blockedDeferred = defer ( ) ; nodeWritable . once ( "drain" , blockedDeferred . resolve ) ; return blockedDeferred . promise ; } } , ( endError ? ) : Promise < void > => { if ( endError ) { handleTsStreamError ( endError ) ; } nodeWritable . end ( ) ; return endDeferred . promise ; } , handleTsStreamError ) ; }