Composition

A composition() allows you to write your own custom agents:
  /*!*/import { composition, pin, map, pipe, state, check, expr, sampler } from '@connectv/core';
import { debounceTime, delay } from 'rxjs/operators';

//
// --> this composition will count consecutive clicks and also determine
// --> if it was a single or multi click
//
/*!*/export const clickCounter = composition(track => {
/*!*/  let click = pin(), reset = pin();                  //--> inputs
/*!*/  let out = pin(), single = pin(), multi = pin();    //--> outputs

  let count = state(0); //--> a state for click count
/*!*/  track(count);         //--> track it so its cleared later on      

  let incr = expr(x => x + 1); //--> this increases the count
  let docheck = sampler();     //--> this determines when should the
                               //... "single v multi" check happen

  click.to(incr.control);      //--> increase only when a click happens

  count.to(incr).to(count)     //--> increase the count and store it
        .to(out, docheck);     //--> also send the result to output `out`
                               //... and the "single v multi" check sampler
                               
  reset.to(map(() => 0)).to(count); //--> set the count to 0 in case of reset

  //
  // --> when the sampler allows the click-count through,
  // ... check if the count is bigger than one,
  // ... if so, send it to the `multi` output,
  // ... if not, send it to the `single` output.
  //
  docheck.to(check(x => x > 1))
    .serialTo(multi, single);

  //
  // --> 200ms after a click without another click,
  // ... commence the "single v multi" check (signal `docheck.control`)
  // ... also reset the counter (with a 1 ms delay)
  //
  click.to(pipe(debounceTime(200)))
    .to(
      docheck.control,
      reset.from(pipe(delay(1)))
    );

/*!*/  return [{ click, reset }, { out, single, multi }]; //--> return the inputs and the outputs
});
Which you can subsequently re-use in other flows:
  import { wrap, map, filter } from '@connectv/core';
import { fromEvent } from 'rxjs';

/*!*/import { clickCounter } from './click-counter';

let p = document.getElementById('p');

wrap(fromEvent(document, 'click'))    //--> get the click event
/*!*/.serialTo(clickCounter())             //--> send it to click-counter
.serialTo(
  filter(x => x > 0),                 //--> the `out` output goes here
  map(() => 'single click'),          //--> the `single` output goes here
  map(() => 'MULTI CLICK!')           //--> the `multi` output goes here
)
.subscribe(v => p.innerHTML = v);

Anatomy

So each composition() has the following form:
import { composition, pin } from '@connectv/core';

let C = composition(track => {
  let inputPinA = pin(), inputPinB = pin(), ...   //--> define input pins
  let outputPinA = pin(), outputPinB = pin(), ... //--> define output pins

  // connect everything to each other

  return [
    { inputPinA, inputPinB, ... },    //--> return input pins
    { outputPinA, outputPinB, ... }   //--> return output pins
  ]
});
Essentially, this is the simplest form of describing any partial reactive flow, i.e. a set of input pins, a set of output pins, and how they are connected to each other.

You would re-use this partial flow in other reactive flows simply like this:
somePins
/*!*/.serialTo(C())
.serialTo(someOtherPinA, someOtherPinB, ...)
As you can C() is basically a factory function that creates instances of your described partial flow.

All composition()s support implicit connections, with the entry pins being the inputs and the exit pins being the outputs, in the provided order. You can of-course connect to them explicitly as well:
let c = C();

somePin.to(c.in('inputPinA'));
c.out('outputPinB').to(someOtherPinB);
...


Tracking

Take another look at this bit in the first example:
/*!*/export const clickCounter = composition(track => {
  let click = pin(), reset = pin();                  //--> inputs
  let out = pin(), single = pin(), multi = pin();    //--> outputs

  let count = state(0); //--> a state for click count
/*!*/  track(count);         //--> track it so its cleared later on
The track() function here registers the state count with the resulting composition. This means that when .clear() is invoked on the composition, it will also be invoked on count, and when .bind() is invoked on the composition, it will also be invoked on count.

The track() function basically syncs life-cycle of internal pins and agents of the composition to the composition instance itself. It is recommended to track() states, sinks, and more complex agents of your compositions.


Construction purity

The factory function you provide composition() will be invoked at least once to determine the signature of the resulting agent. This signature will be used as the signature for all future instances, so you cannot change it later on. This also means that your factory function must be pure, i.e. the state of the program should not change by invoking it.

If you need dynamic signatures, for example based on some parameter, you can utilize the following pattern:
let C = param => composition(track => {
  ...

  return [inputsBasedOn(param), outputsBasedOn(param)];
})();
If your factory function needs to be impure, then you can provide composition() with a signature as its first argument. This way your factory will only be called when an instance is created:
let C = composition({
  inputs: ['a', 'b', 'c'],
  outputs: ['d', 'e']
}, track => {
  ...

  return [{a, b, c}, {d, e}]
})


Further reading




Copied to Clipboard!