import { _ } from 'streamline-runtime' ; import * as generic from './generic' ; import * as stopException from '../stop-exception' ; import { Reader } from '../reader' ; import { Writer } from '../writer' ; import { nextTick } from '../util' ; var lastId = 0 ; var tracer ; export interface Uturn < T > { reader : Reader < T > ; writer : Writer < T > ; end : ( _ : _ ) => void ; } export function create < T > ( ) { var state = 'idle' , pendingData , error ; const id = ++ lastId ; var pendingReaderCb ; function bounceReader ( err ? , val ? ) { const lcb = pendingReaderCb ; pendingReaderCb = null ; if ( lcb ) lcb ( err , val ) ; } var pendingWriterCb ; function bounceWriter ( err ? , val ? ) { const lcb = pendingWriterCb ; pendingWriterCb = null ; if ( lcb ) lcb ( err , val ) ; } var pendingStopCb ; function bounceStop ( err ? , val ? ) { const lcb = pendingStopCb ; pendingStopCb = null ; if ( lcb ) lcb ( err , val ) ; } const uturn = { reader : new Reader ( _ . cast ( function read ( cb ) { nextTick ( ( ) => { tracer && tracer ( id , "READ" , state , pendingData ) ; const st = state ; switch ( st ) { case 'writing' : state = pendingData === undefined ? 'done' : 'idle' ; bounceWriter ( null , uturn . writer ) ; cb ( null , pendingData ) ; pendingData = undefined ; break ; case 'idle' : state = 'reading' ; pendingReaderCb = cb ; break ; case 'readStopping' : case 'writeStopping' : state = 'done' ; const arg = stopException . unwrap ( error ) ; bounceStop ( ) ; cb ( arg && arg !== true ? arg : null ) ; break ; case 'done' : cb ( error ) ; break ; default : state = 'done' ; cb ( error || new Error ( 'invalid state ' + st ) ) ; break ; } } ) ; } ) , _ . cast ( function stop ( cb , arg ? ) { nextTick ( ( ) => { error = error || stopException . make ( arg ) ; tracer && tracer ( id , "STOP READER" , state , arg ) ; const st = state ; switch ( st ) { case 'reading' : state = 'done' ; bounceReader ( arg && arg !== 1 ? arg : null ) ; cb ( ) ; break ; case 'writing' : state = 'done' ; bounceWriter ( error , uturn . writer ) ; cb ( ) ; break ; case 'idle' : state = 'readStopping' ; pendingStopCb = cb ; break ; case 'done' : cb ( error ) ; break ; default : state = 'done' ; cb ( error || new Error ( 'invalid state ' + st ) ) ; break ; } } ) ; } ) ) , writer : new Writer ( _ . cast ( function write ( this , cb , data ) { nextTick ( ( ) => { tracer && tracer ( id , "WRITE" , state , data ) ; const st = state ; switch ( st ) { case 'reading' : state = data === undefined ? 'done' : 'idle' ; bounceReader ( error , data ) ; cb ( null , this ) ; break ; case 'idle' : state = 'writing' ; pendingWriterCb = cb ; pendingData = data ; break ; case 'readStopping' : state = 'done' ; bounceStop ( ) ; cb ( error ) ; break ; case 'done' : cb ( error || 'invalid state ' + st ) ; break ; default : state = 'done' ; cb ( new Error ( 'invalid state ' + st ) ) ; break ; } } ) ; } ) , _ . cast ( function stop ( cb , arg ? ) { nextTick ( ( ) => { tracer && tracer ( id , "STOP WRITER" , state , arg ) ; error = error || stopException . make ( arg ) ; const st = state ; switch ( st ) { case 'reading' : state = 'done' ; bounceReader ( arg && arg !== 1 ? arg : null ) ; cb ( ) ; break ; case 'idle' : state = 'writeStopping' ; pendingStopCb = cb ; break ; case 'done' : cb ( error ) ; break ; default : state = 'done' ; cb ( new Error ( 'invalid state ' + st ) ) ; break ; } } ) ; } ) ) , end : _ . cast ( function ( err ) { nextTick ( ( ) => { tracer && tracer ( id , "END" , state , err ) ; err = stopException . unwrap ( err ) ; error = error || err ; state = 'done' ; bounceReader ( error ) ; bounceWriter ( error , uturn . writer ) ; bounceStop ( error ) ; } ) ; } ) , } ; return uturn ; }