import { Subscription } from 'rxjs' ; import emission from '../shared/emission' ; import map from '../pin/map' ; import value from '../pin/value' ; import source , { Source } from '../pin/source' ; import { Agent } from './agent' ; export type AgentFactory = ( ) => Agent ; export type ExecResult = { label : string ; value : any ; } ; param factory param sub param unsub param outs export function exec ( factory , sub ? , unsub ? , outs ? ) { return map ( ( data , done , error , context ) => { let _agent = factory ( ) ; let _sources = < { [ input ] : Source ; } > { } ; let _subs = new Subscription ( ) ; let _cleanup = ( ) => { Object . values ( _sources ) . forEach ( s => s . clear ( ) ) ; _agent . clear ( ) ; _subs . unsubscribe ( ) ; if ( unsub ) unsub ( _subs ) ; } ; if ( data ) Object . keys ( data ) . forEach ( ( input ) => _agent . in ( input ) . from ( _sources [ input ] = source ( ) ) ) ; let _outs = _agent . signature . outputs || [ ] ; if ( outs ) _outs = outs ( ) ; _outs . forEach ( ( label ) => { _subs . add ( _agent . out ( label ) . subscribe ( value => { _cleanup ( ) ; done ( { label , value } ) ; } , err => { _cleanup ( ) ; error ( err ) ; } ) ) ; } ) ; if ( sub ) sub ( _subs ) ; if ( data ) Object . entries ( data ) . forEach ( ( [ input , value ] ) => _sources [ input ] . emit ( emission ( value , context ) ) ) ; } ) ; } param factory param data export function call ( factory , data ) { return value ( data ) . to ( exec ( factory ) ) ; } export default call ;