import { tap } from 'rxjs/operators' ; import { Emission , MergedEmissionContextVal } from '../shared/emission' ; import group from '../pin/group' ; import pin from '../pin/pin' ; import { Source } from '../pin/source' ; import { block } from '../pin/filter' ; import pipe from '../pin/pipe' ; import { Agent } from './agent' ; type _CacheType = { [ key ] : Emission ; } ; /**
 *
 * Represents [join](https://connective.dev/docs/join) agents.
 *
 */ export class Join extends Agent { private _inject : Source ; private _cache : { [ fork ] : _CacheType ; } = { } ; param keys param pop constructor ( readonly keys , readonly pop = true ) { super ( { inputs : keys , outputs : [ 'output' ] } ) ; this . _inject = new Source ( ) ; } private _receive ( key , emission ) { if ( emission . context . __fork ) { if ( emission . context . __fork instanceof MergedEmissionContextVal ) emission . context . __fork . values . forEach ( v => this . _fill ( v , key , emission ) ) ; else this . _fill ( emission . context . __fork , key , emission ) ; } else this . _fill ( [ ] , key , emission ) ; } private _cache_key ( fork ) { return fork . join ( ';' ) ; } private _fill ( fork , key , emission ) { let _fork = this . _cache_key ( fork ) ; let _cache = this . _cache [ _fork ] = this . _cache [ _fork ] || { } ; _cache [ key ] = emission ; if ( this . _complete ( _cache ) ) this . _emit ( _cache , fork ) ; } private _emit ( cache , fork ) { delete this . _cache [ this . _cache_key ( fork ) ] ; let emission = Emission . from ( Object . values ( cache ) , Object . entries ( cache ) . reduce ( ( obj , entry ) => { obj [ entry [ 0 ] ] = entry [ 1 ] . value ; return obj ; } , < { [ key ] : any ; } > { } ) ) ; emission . context . __fork = this . pop ? fork . slice ( 0 , - 1 ) : [ ... fork ] ; this . _inject . emit ( emission ) ; } private _complete ( cache ) { return Object . values ( cache ) . length == this . keys . length ; } protected createOutput ( label ) { this . checkOutput ( label ) ; return group ( group ( ... this . keys . map ( key => this . in ( key ) . to ( pipe ( tap ( e => this . _receive ( key , e ) ) ) ) ) ) . to ( block ( ) ) , this . _inject ) . to ( pin ( ) ) ; } protected createEntries ( ) { return this . keys . map ( key => this . in ( key ) ) ; } protected createExits ( ) { return [ this . output ] ; } /**
     *
     * Shortcut for `.out('output')`, which will emit the joined object.
     * [Read this](https://connective.dev/docs/handle-error#signature) for more details.
     *
     */ public get output ( ) { return this . out ( 'output' ) ; } public clear ( ) { this . _inject . clear ( ) ; this . _cache = { } ; return this ; } } param keys export function join ( ... keys ) { return new Join ( keys ) ; } param keys export function peekJoin ( ... keys ) { return new Join ( keys , false ) ; } export default join ;