const fibers = require ( 'fibers' ) ; export type Callback < T > = ( err : any , result ? : T ) => void ; export type Thunk < T > = ( cb : Callback < T > ) => void ; export let wait = < T = any > ( promiseOrCallback ) : T => { const fiber = fibers . current ; if ( ! fiber ) throw new Error ( 'cannot wait: no fiber' ) ; if ( typeof promiseOrCallback === 'function' ) { promiseOrCallback ( ( err , res ) => { process . nextTick ( ( ) => { let cx = globals . context ; try { if ( err ) { fiber . throwInto ( err ) ; } else { fiber . run ( res ) ; } } finally { globals . context = cx ; cx = null ; } } ) ; } ) ; } else { promiseOrCallback . then ( res => { let cx = globals . context ; try { fiber . run ( res ) ; } finally { globals . context = cx ; cx = null ; } } ) . catch ( e => { let cx = globals . context ; try { fiber . throwInto ( e ) ; } finally { globals . context = cx ; cx = null ; } } ) ; } let cx = globals . context ; try { return fibers . yield ( ) ; } catch ( e ) { throw ( fullStackError && fullStackError ( e ) ) || e ; } finally { globals . context = cx ; cx = null ; } } ; export let run = < T > ( fn ) : Promise < T > => { if ( typeof fn !== 'function' ) { throw new Error ( 'run() should take a function as argument' ) ; } return new Promise ( ( resolve , reject ) => { const cx = globals . context ; fibers ( ( ) => { try { resolve ( fn ( ) ) ; } catch ( e ) { reject ( ( cleanFiberStack && cleanFiberStack ( e ) ) || e ) ; } } ) . run ( ) ; globals . context = cx ; } ) ; } ; export function funnel ( max = - 1 ) { if ( typeof max !== 'number' ) { throw new Error ( 'bad max number: ' + max ) ; } const _max = max === 0 ? exports . funnel . defaultSize : max ; let queue = [ ] ; let active = 0 ; let closed = false ; function tryEnter < T > ( fn ) { if ( active < _max ) { active ++ ; try { return fn ( ) ; } finally { active -- ; const hk = queue . shift ( ) ; if ( hk ) { hk . notify ( ) ; } } } else { return overflow < T > ( fn ) ; } } function overflow < T > ( fn ) { const hk = handshake ( ) ; queue . push ( hk ) ; hk . wait ( ) ; if ( closed ) { throw new Error ( `cannot execute: funnel has been closed` ) ; } return tryEnter < T > ( fn ) ; } const fun = function < T > ( fn ) : T { if ( closed ) { throw new Error ( `cannot execute: funnel has been closed` ) ; } if ( _max < 0 || _max === Infinity ) { return fn ( ) ; } return tryEnter ( fn ) ; } as Funnel ; fun . close = ( ) => { queue . forEach ( hk => { hk . notify ( ) ; } ) ; queue = [ ] ; closed = true ; } ; return fun ; } ( funnel as any ) . defaultSize = 4 ; export interface Funnel { < T > ( fn ) : T ; close ( ) : void ; } export function handshake < T = void > ( ) { let callback = undefined , notified = false ; return { wait ( ) { return wait < T > ( ( cb ) => { if ( callback ) throw new Error ( 'already waiting' ) ; if ( notified ) setImmediate ( cb ) ; else callback = cb ; notified = false ; } ) ; } , notify ( ) { if ( ! callback ) notified = true ; else setImmediate ( callback ) ; callback = undefined ; } , } ; } export interface Handshake < T = void > { wait ( ) : void ; notify ( ) : void ; } export interface QueueOptions { max ? : number ; } export class Queue < T > { _max : number ; _callback : Callback < T > | undefined ; _err : any ; _q : ( T | undefined ) [ ] = [ ] ; _pendingWrites : [ Callback < T > , T | undefined ] [ ] = [ ] ; constructor ( options ? ) { if ( typeof options === 'number' ) { options = { max : options , } ; } options = options || { } ; this . _max = options . max != null ? options . max : - 1 ; } read ( ) { return wait < T > ( ( cb ) => { if ( this . _callback ) throw new Error ( 'already getting' ) ; if ( this . _q . length > 0 ) { const item = this . _q . shift ( ) ; if ( this . _q . length === 0 ) this . _q = [ ] ; setImmediate ( ( ) => { cb ( this . _err , item ) ; } ) ; if ( this . _pendingWrites . length > 0 ) { const wr = this . _pendingWrites . shift ( ) ; setImmediate ( ( ) => { wr && wr [ 0 ] ( this . _err , wr [ 1 ] ) ; } ) ; } } else { this . _callback = cb ; } } ) ; } write ( item ) { return wait < T > ( ( cb ) => { if ( this . put ( item ) ) { setImmediate ( ( ) => { cb ( this . _err ) ; } ) ; } else { this . _pendingWrites . push ( [ cb , item ] ) ; } } ) ; } put ( item , force ? ) { if ( ! this . _callback ) { if ( this . _max >= 0 && this . _q . length >= this . _max && ! force ) return false ; this . _q . push ( item ) ; } else { const cb = this . _callback ; this . _callback = undefined ; setImmediate ( ( ) => { cb ( this . _err , item ) ; } ) ; } return true ; } end ( ) { this . put ( undefined , true ) ; } peek ( ) { return this . _q [ 0 ] ; } contents ( ) { return this . _q . slice ( 0 ) ; } adjust ( fn ) { const nq = fn . call ( null , this . _q ) ; if ( ! Array . isArray ( nq ) ) throw new Error ( 'adjust function does not return array' ) ; this . _q = nq ; } get length ( ) { return this . _q . length ; } } export function withContext < T > ( fn , cx ) { if ( ! fibers . current ) throw new Error ( 'withContext(fn) not allowed outside run()' ) ; const oldContext = globals . context ; globals . context = cx || Object . create ( oldContext ) ; try { return fn ( ) ; } finally { globals . context = oldContext ; } } export function context < T = any > ( ) { return globals . context ; } export function map < T , R > ( collection , fn ) { return wait ( Promise . all ( collection . map ( item => { return run ( ( ) => fn ( item ) ) ; } ) ) ) ; } export function sleep ( n ) { wait ( cb => setTimeout ( cb , n ) ) ; } export function canWait ( ) { return ! ! fibers . current ; } export function eventHandler < T extends Function > ( handler ) { const wrapped = function ( this , ... args ) { if ( canWait ( ) ) { handler . apply ( this , args ) ; } else { run ( ( ) => withContext ( ( ) => handler . apply ( this , args ) , { } ) ) . catch ( err => { console . error ( err ) ; } ) ; } } as any ; Object . defineProperty ( wrapped , 'length' , { value : handler . length } ) ; return wrapped ; } declare const global ; const secret = '_20c7abceb95c4eb88b7ca1895b1170d1' ; const globals = ( global [ secret ] = global [ secret ] || { context : { } } ) ; let fullStackError ; let cleanFiberStack ; let cannotOverrideStackWarned = false ; function overrideStack ( e , getFn ) { try { Object . defineProperty ( e , 'stack' , { get : getFn , enumerable : true , configurable : true , } ) ; } catch ( e ) { if ( ! cannotOverrideStackWarned ) { console . warn ( `[F-PROMISE] attempt to override e.stack failed (warning will not be repeated)` ) ; cannotOverrideStackWarned = true ; } } } if ( process . env . FPROMISE_STACK_TRACES === 'whole' ) { fullStackError = function fullStackError ( e ) { if ( ! ( e instanceof Error ) ) { return e ; } const localError = new Error ( '__fpromise' ) ; const fiberStack = e . stack || '' ; overrideStack ( e , function ( ) { const localStack = localError ? localError . stack || '' : '' ; return fiberStack + localStack ; } ) ; return e ; } ; } else if ( process . env . FPROMISE_STACK_TRACES !== 'fast' ) { fullStackError = function fullStackError ( e ) { if ( ! ( e instanceof Error ) ) { return e ; } const localError = new Error ( '__f-promise' ) ; const fiberStack = e . stack || '' ; overrideStack ( e , function ( ) { const localStack = localError ? localError . stack || '' : '' ; return ( fiberStack + '\n' + localStack . split ( '\n' ) . slice ( 1 ) . filter ( line => { return ! /\/f-promise\// . test ( line ) ; } ) . join ( '\n' ) ) ; } ) ; return e ; } ; cleanFiberStack = function cleanFiberStack ( e ) { if ( ! ( e instanceof Error ) ) { return e ; } const fiberStack = e . stack || '' ; overrideStack ( e , function ( ) { return fiberStack . split ( '\n' ) . filter ( line => { return ! /\/f-promise\// . test ( line ) ; } ) . join ( '\n' ) ; } ) ; return e ; } ; } if ( process . execArgv . find ( str => str . startsWith ( '--inspect-brk' ) ) ) { const isDebugEval = ( ) => ( new Error ( ) . stack || '' ) . indexOf ( '.remoteFunction (<anonymous>' ) >= 0 ; const oldWait = wait ; const flushDelayed = ( ) => { if ( fibers . current . delayed ) { const delayed = fibers . current . delayed ; fibers . current . delayed = undefined ; for ( const arg of delayed ) { try { oldWait ( arg ) ; } catch ( err ) { console . error ( `delayed 'wait' call failed: ${ err . message }` ) ; } } } } ; wait = < T > ( arg ) : T => { if ( isDebugEval ( ) ) { if ( ! fibers . current . delayed ) fibers . current . delayed = [ ] ; fibers . current . delayed . push ( arg ) ; throw 'would yield' ; } flushDelayed ( ) ; return oldWait ( arg ) ; } ; const oldRun = run ; run = < T > ( fn ) : Promise < T > => { if ( isDebugEval ( ) ) throw 'would start a fiber' ; else return oldRun ( fn ) ; } ; console . log ( 'Running with f-promise debugger hooks' ) ; }