"use strict" ; /**
 * 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 { Reader , ParallelOptions } from "./reader" ; import { create as createUturn } from './devices/uturn' ; import * as nodeStream from "stream" ; import * as sys from 'util' ; export class Writer < T > { write : ( this : Writer < T > , _ : _ , value ? : T ) => this ; ended : boolean ; constructor ( write , stop ? ) { if ( typeof write !== 'function' ) throw new Error ( "invalid writer.write: " + ( write && typeof write ) ) ; this . ended = false ; this . write = ( _ , data ) => { if ( data === undefined ) { if ( ! this . ended ) write . call ( this , _ ) ; this . ended = true ; } else { if ( this . ended ) throw new Error ( "invalid attempt to write after end" ) ; write . call ( this , _ , data ) ; } return this ; } ; if ( stop ) { this . stop = ( _ , arg ? ) => { stop . call ( this , _ , arg ) ; return this ; } ; } } writeAll ( _ , val ) { this . write ( _ , val ) ; this . write ( _ , undefined ) ; return this ; } ; stop ( _ , arg ? ) { this . write ( _ , undefined ) ; return this ; } ; end ( ) { if ( arguments . length > 0 ) throw new Error ( "invalid end call: " + arguments . length + " arg(s)" ) ; _ . run ( _ => this . write ( _ , undefined ) ) ; return this ; } ; get pre ( ) : Pre < T > { return new PreImpl ( this ) as Pre < T > ; } nodify ( ) { const self = this ; const stream = new nodeStream . Writable ( ) ; const anyStream = stream ; anyStream . _write = function ( chunk , encoding ? , done ? ) { if ( chunk && encoding && encoding !== 'buffer' ) chunk = chunk . toString ( encoding ) ; _ . run ( _ => self . write ( _ , chunk ) , err => { if ( err ) return stream . emit ( 'error' , err ) as never ; if ( done ) done ( ) ; } ) ; } ; const end = stream . end ; anyStream . end = function ( chunk , encoding ? , cb ? ) { end . call ( stream , chunk , encoding , ( err ) => { if ( err ) return stream . emit ( 'error' , err ) as never ; cb = cb || ( ( err ) => { } ) ; _ . run ( _ => self . write ( _ , undefined ) , cb ) ; } ) ; } ; return stream ; } get result ( ) : any { throw new Error ( "result not supported" ) ; } } ; export function create < T > ( write , stop ? ) { return new Writer ( write , stop ) ; } exports . decorate = function ( proto ) { const writerProto = Writer . prototype ; Object . getOwnPropertyNames ( Writer . prototype ) . forEach ( k => { if ( k == 'constructor' || k == 'result' ) return ; if ( k == 'pre' ) { Object . defineProperty ( proto , k , { get ( this ) { return new PreImpl ( this ) ; } } ) ; } else { if ( ! proto [ k ] ) proto [ k ] = writerProto [ k ] ; } } ) ; return proto ; } ; export class PreImpl < T > { writer : Writer < T > ; constructor ( writer ) { if ( typeof writer . write !== 'function' ) throw new Error ( "invalid pre writer: " + sys . inspect ( writer ) ) ; this . writer = writer ; } } export interface Pre < T > extends PreImpl < T > { map < U > ( fn , thisObj ? ) : Writer < U > ; tee ( writer ) : Writer < T > ; concat ( readers ) : Writer < T > ; transform < U > ( fn , thisObj ? ) : Writer < U > ; filter ( fn , thisObj ? ) : Writer < T > ; until ( fn , thisObj ? ) : Writer < T > ; while ( fn , thisObj ? ) : Writer < T > ; limit ( n , stopArg ? ) : Writer < T > ; skip ( n ) : Writer < T > ; parallel ( options , consumer ) : Writer < T > ; buffer ( max ) : Writer < T > ; nodeTransform < U > ( duplex ) : Writer < U > ; } process . nextTick ( ( ) => { const preProto = PreImpl . prototype ; const api = Reader . prototype ; [ 'map' , 'tee' , 'concat' , 'transform' , 'filter' , 'until' , 'while' , 'limit' , 'skip' , 'parallel' , 'buffer' , 'nodeTransform' ] . forEach ( ( name ) => { preProto [ name ] = function ( this , arg ) { const uturn = require ( './devices/uturn' ) . create ( ) ; uturn . reader [ name ] ( arg ) . pipe ( uturn . end , this . writer ) ; return uturn . writer ; } ; } ) ; } ) ;