import * as stream from "stream" ; import { callbackify } from "util" ; import { ClientProxy , ClientServerProxy } from "../common/proxy" ; import { isPromise } from "../common/util" ; import { DuplexProxy , ReadableProxy , WritableProxy } from "../server/stream" ; export interface ClientWritableProxy extends WritableProxy , ClientServerProxy < stream . Writable > { } export class Writable < T extends ClientWritableProxy = ClientWritableProxy > extends ClientProxy < T > implements stream . Writable { public get writable ( ) : boolean { throw new Error ( "not implemented" ) ; } public get writableFinished ( ) : boolean { throw new Error ( "not implemented" ) ; } public get writableHighWaterMark ( ) : number { throw new Error ( "not implemented" ) ; } public get writableLength ( ) : number { throw new Error ( "not implemented" ) ; } public _write ( ) { throw new Error ( "not implemented" ) ; } public _destroy ( ) { throw new Error ( "not implemented" ) ; } public _final ( ) { throw new Error ( "not implemented" ) ; } public pipe < T > ( ) { throw new Error ( "not implemented" ) ; } public cork ( ) { throw new Error ( "not implemented" ) ; } public uncork ( ) { throw new Error ( "not implemented" ) ; } public destroy ( ) { this . catch ( this . proxy . destroy ( ) ) ; } public setDefaultEncoding ( encoding ) { return this . catch ( this . proxy . setDefaultEncoding ( encoding ) ) ; } public write ( chunk , encoding ? , callback ? ) { if ( typeof encoding === "function" ) { callback = encoding ; encoding = undefined ; } callbackify ( this . proxy . write ) ( chunk , encoding , ( error ) => { if ( callback ) { callback ( error ) ; } } ) ; return true ; } public end ( data ? , encoding ? , callback ? ) { if ( typeof data === "function" ) { callback = data ; data = undefined ; } if ( typeof encoding === "function" ) { callback = encoding ; encoding = undefined ; } callbackify ( this . proxy . end ) ( data , encoding , ( ) => { if ( callback ) { callback ( ) ; } } ) ; } protected handleDisconnect ( ) { this . emit ( "close" ) ; this . emit ( "finish" ) ; } } export interface ClientReadableProxy extends ReadableProxy , ClientServerProxy < stream . Readable > { } export class Readable < T extends ClientReadableProxy = ClientReadableProxy > extends ClientProxy < T > implements stream . Readable { public get readable ( ) : boolean { throw new Error ( "not implemented" ) ; } public get readableHighWaterMark ( ) : number { throw new Error ( "not implemented" ) ; } public get readableLength ( ) : number { throw new Error ( "not implemented" ) ; } public _read ( ) { throw new Error ( "not implemented" ) ; } public read ( ) { throw new Error ( "not implemented" ) ; } public _destroy ( ) { throw new Error ( "not implemented" ) ; } public unpipe ( ) { throw new Error ( "not implemented" ) ; } public pause ( ) { throw new Error ( "not implemented" ) ; } public resume ( ) { throw new Error ( "not implemented" ) ; } public isPaused ( ) { throw new Error ( "not implemented" ) ; } public wrap ( ) { throw new Error ( "not implemented" ) ; } public push ( ) { throw new Error ( "not implemented" ) ; } public unshift ( ) { throw new Error ( "not implemented" ) ; } public pipe < P extends NodeJS . WritableStream > ( destination , options ? ) { const writableProxy = ( ( destination as any ) as Writable ) . proxyPromise ; if ( ! writableProxy ) { throw new Error ( "can only pipe stream proxies" ) ; } this . catch ( isPromise ( writableProxy ) ? writableProxy . then ( ( p ) => this . proxy . pipe ( p , options ) ) : this . proxy . pipe ( writableProxy , options ) ) ; return destination ; } public [ Symbol . asyncIterator ] ( ) { throw new Error ( "not implemented" ) ; } public destroy ( ) { this . catch ( this . proxy . destroy ( ) ) ; } public setEncoding ( encoding ) { return this . catch ( this . proxy . setEncoding ( encoding ) ) ; } protected handleDisconnect ( ) { this . emit ( "close" ) ; this . emit ( "end" ) ; } } export interface ClientDuplexProxy extends DuplexProxy , ClientServerProxy < stream . Duplex > { } export class Duplex < T extends ClientDuplexProxy = ClientDuplexProxy > extends Writable < T > implements stream . Duplex , stream . Readable { private readonly _readable : Readable ; public constructor ( proxyPromise ) { super ( proxyPromise ) ; this . _readable = new Readable ( proxyPromise , false ) ; } public get readable ( ) : boolean { return this . _readable . readable ; } public get readableHighWaterMark ( ) : number { return this . _readable . readableHighWaterMark ; } public get readableLength ( ) : number { return this . _readable . readableLength ; } public _read ( ) { this . _readable . _read ( ) ; } public read ( ) { this . _readable . read ( ) ; } public unpipe ( ) { this . _readable . unpipe ( ) ; return this ; } public pause ( ) { this . _readable . unpipe ( ) ; return this ; } public resume ( ) { this . _readable . resume ( ) ; return this ; } public isPaused ( ) { return this . _readable . isPaused ( ) ; } public wrap ( ) { this . _readable . wrap ( ) ; return this ; } public push ( ) { return this . _readable . push ( ) ; } public unshift ( ) { this . _readable . unshift ( ) ; } public [ Symbol . asyncIterator ] ( ) { return this . _readable [ Symbol . asyncIterator ] ( ) ; } public setEncoding ( encoding ) { return this . catch ( this . proxy . setEncoding ( encoding ) ) ; } protected handleDisconnect ( ) { super . handleDisconnect ( ) ; this . emit ( "end" ) ; } }