/**
 * Promise-based object stream with seamless support for back-pressure and error
 * handling, written in Typescript.
 *
 * Copyright (C) 2015 Martin Poelstra
 * License: MIT
 */ import { Readable , Stream , Writable } from "./Stream" ; import { TrackedVoidPromise , track , swallowErrors , noop } from "./util" ; import { BatcherOptions } from "./transformers" ; export type Transform < In , Out > = ( readable : Readable < In > , writable : Writable < Out > ) => void ; export function compose < In , Middle , Out > ( t1 , t2 ) { return ( readable , writable ) : void => { const stream = new Stream < Middle > ( ) ; t1 ( readable , stream ) ; t2 ( stream , writable ) ; } ; } function composeEnders ( ender , defaultEnder ) { if ( ! ender ) { return defaultEnder ; } return ( error ? ) => { return Promise . resolve ( error ) . then ( ender ) . then ( ( ) => defaultEnder ( error ) , ( enderError ) => { return Promise . resolve ( defaultEnder ( error ?? enderError ) ) . then ( ( ) => Promise . reject ( enderError ) , ( ) => Promise . reject ( enderError ) ) ; } ) ; } ; } export function map < T , R > ( readable , writable , mapper , ender ? , aborter ? ) { writable . aborted ( ) . catch ( ( err ) => readable . abort ( err ) ) ; readable . aborted ( ) . catch ( ( err ) => writable . abort ( err ) ) ; readable . forEach ( ( v ) => writable . write ( mapper ( v ) ) , composeEnders ( ender , ( error ? ) => writable . end ( error , readable . result ( ) ) ) , aborter ) ; } export function filter < T > ( readable , writable , filterer , ender ? , aborter ? ) { writable . aborted ( ) . catch ( ( err ) => readable . abort ( err ) ) ; readable . aborted ( ) . catch ( ( err ) => writable . abort ( err ) ) ; readable . forEach ( ( v ) : void | Promise < void > => { const b = filterer ( v ) ; if ( ! b ) { return ; } else if ( b === true ) { return writable . write ( v ) ; } else { return Promise . resolve ( b ) . then ( ( resolvedB ) => { if ( resolvedB ) { return writable . write ( v ) ; } } ) ; } } , composeEnders ( ender , ( error ? ) => writable . end ( error , readable . result ( ) ) ) , aborter ) ; } export function batch < T > ( readable , writable , maxBatchSize , { minBatchSize = maxBatchSize , flushTimeout , handleError , } = { } ) { writable . aborted ( ) . catch ( ( err ) => readable . abort ( err ) ) ; readable . aborted ( ) . catch ( ( err ) => writable . abort ( err ) ) ; let queue = [ ] ; let pendingWrite ; let timeout ; async function flush ( ) { if ( queue . length ) { const peeled = queue ; queue = [ ] ; await writable . write ( peeled ) . catch ( handleError && ( ( e ) => handleError ( e , peeled ) ) ) ; } } async function earlyFlush ( ) { while ( queue . length >= minBatchSize && queue . length < maxBatchSize ) { pendingWrite = track ( flush ( ) ) ; swallowErrors ( pendingWrite . promise ) ; await pendingWrite . promise ; } pendingWrite = undefined ; } function clearFlushTimeout ( ) { if ( timeout !== undefined ) { clearTimeout ( timeout ) ; timeout = undefined ; } } function startFlushTimeout ( ) { if ( typeof flushTimeout === "number" ) { clearFlushTimeout ( ) ; timeout = setTimeout ( ( ) => { if ( ! pendingWrite ?. isPending && queue . length > 0 ) { pendingWrite = track ( flush ( ) ) ; swallowErrors ( pendingWrite . promise ) ; } } , flushTimeout ) ; } } function consumeEarlyFlushError ( ) { if ( pendingWrite ?. isRejected ) { const reason = pendingWrite . reason as Error ; pendingWrite = undefined ; return reason ; } } async function settleEarlyFlush ( ) { try { await pendingWrite ?. promise ; } catch ( e ) { return consumeEarlyFlushError ( ) ; } } function throwIfThrowable ( e ) { if ( e ) { throw e ; } } readable . forEach ( async ( v ) : Promise < void > => { startFlushTimeout ( ) ; queue . push ( v ) ; let flushFailureError ; let earlyFlushError ; if ( queue . length >= maxBatchSize ) { try { earlyFlushError = await settleEarlyFlush ( ) ; await flush ( ) ; } catch ( e ) { flushFailureError = e ; } } else if ( ! pendingWrite ) { swallowErrors ( earlyFlush ( ) ) ; } const toThrow = earlyFlushError ?? consumeEarlyFlushError ( ) ?? flushFailureError ; throwIfThrowable ( toThrow ) ; } , async ( error ? ) => { clearFlushTimeout ( ) ; let flushError ; const earlyFlushError = await settleEarlyFlush ( ) ; try { await flush ( ) ; } catch ( e ) { flushError = e ; } const toThrow = earlyFlushError ?? flushError ; await writable . end ( error , readable . result ( ) ) ; throwIfThrowable ( toThrow ) ; } , ( ) => { if ( ! pendingWrite ) { swallowErrors ( earlyFlush ( ) ) ; } } ) ; } export default Transform ;