Introduction

┌─┐┌─┐┬  ┬  ┌─┐┬─┐ ┬┌─┐ ┌─┐┌─┐┌┬┐
│  │ ││  │  ├─┤├┬┘ │└─┐ │  │ ││││
└─┘└─┘┴─┘┴─┘┴ ┴┴└─└┘└─┘o└─┘└─┘┴ ┴

version 0.6.x

Collar.js helps you turn your thoughts into code, and visualize your thoughts in order to help you make high quality software.

Installation

install collar.js with npm

npm install collar.js --save

install collar.js-dev-client with npm

npm install collar.js-dev-client --save-dev

install collar-dev-server with npm

npm install collar-dev-server -g

Collar.js works on both font end and back end.

node.js

use npm to install collar.js

browser

options 1 : use npm to install collar.js and package it with webpack or similar tools
options 2 : directly download the latest version from http://collarjs.com and include it in your web page

install collar dev client

use npm to install collar.js-dev-client

or download the latest version from http://collarjs.com and include it in your web page

install collar dev server

use npm to install collar-dev-server

Usage

require collar.js and create a namespace

const collar = require("collar.js");
const ns = collar.ns("com.collarjs.demo");

// for browser usage :
// if you download collar.min.js and include it in your page
// collar is already registered as a global variable
const ns = collar.ns("com.collarjs.demo");

record your thoughts with collar.js API

const uiSensor = ns.sensor("get user input");

var registerPipeline = uiSensor
  .filter("when user click 'register' button")
  .do("get user email and password from UI")
  .map("prepare 'register' event")
  .do("check email and password pair in database")

registerPipeline
  .filter("when register succeeds")
  .do("show home view")

registerPipeline
  .filter("when register fails")
  .do("show error view")

require collar.js and create a namespace
record your thoughts with collar.js API
implement your thoughts by implementing the nodes in your flow
visualize your thoughts and data flow with collar-dev-server

The example shown in this section is visualized as following:

example flow

Single Responsibility Principle

every module or class should have responsibility over a single part of the functionality provided by the software, and that responsibility should be entirely encapsulated by the class. All its services should be narrowly aligned with that responsibility

In collar.js, the responsibility is divided into four major ones:

signal source, or sensor, which gets information from external system and emits signal to internal system
signal filter, which allows signals to pass under certain conditions
actuator, which makes side effects according to the incoming signal. An actuator only makes side effects, it does not change the incoming signal. The same signal will be emitted by actuator
processor, which makes no side effect, but changes the incoming signal, and emits the new signal

General system architecture

With the 4 responsibilities, you can construct a general message oriented system :

environment --> [sensor] -- input -> [filter] -> [actuator] -> [processor] --> output

The system contains a filter-actuator-processor pipeline: the filter decides which input to process, and blocks unexpected inputs. Actuator makes side effects, it could be accessing database, printing a log, making an http request, or updating the UI view. Finally, a processor processes the input and generate the output of your system.

The input could be the output of another system or directly from the environment. As the environment might not generate the input with right data structure, a sensor is used to transform the environment signal to your domain data.

Core API

sensor ( comment, inputs, outputs, watch )

Create a sensor

const ns = collar.ns('collarjs.demo.sensor');
const sensor = ns.sensor(function(options) {
  setTimeout(() => {
    this.send({
      event : "timeout",
      delay : 10000
    })
  }, 10000);
});

// sensor starts to watch when it is created

Create a sensor with options

const sensor = ns.sensor(function(options) {
  var delay = 10000; // default delay
  if (options.delay)
    delay = options.delay;  // use delay in options
  else
    return; // if no delay defined, watch nothing

  setTimeout(() => {
    this.send({
      event : "timeout",
      delay : delay
    })
  }, delay);
});

sensor.watch({delay : 5000}); // start watch with options

Sensor does not accept incoming signals, it watches the external world (compared to your system) and send signals to your system. It is the signal source.

Signature

#NS.sensor(comment : string, inputs : map, outputs : map, watch : function) : Node

Parameters

Parameter	Type	Description
comment	string	optional, the comment of the sensor
inputs	map	optional, the input description
outputs	map	optional, the output description
watch	function	the watch function

function watch(options : any) : void

Pass a watch function as parameter to watch the external world

filter ( comment, inputs, outputs, accept )

Create a filter

// only allow the signals, which have the 'age' greater than 18, pass
const ns = collar.ns('collarjs.demo.filter');
const filter = ns.filter(signal => {
  return signal.get("age") > 18;
});

Create a placeholder filter to record your thoughts. In the following example, neither the filter nor the actuator is implemented, but it is a valid data flow, which correctly records your thoughts. When you run the following code, it acts like a pass-through flow.

// in collar dev tool, these two nodes are in gray color,
// to warning you that they are not implemented yet.
ns
  .filter("when age > 18")
  .actuator("show alcohol products");

Control signal flow by applying a filter to it.

Signature

#NS.filter(comment : string, inputs : map, outputs : map, accept : function) : Node

Alias

when()

Parameters

Parameter	Type	Description
comment	string	optional, the comment of the filter
inputs	map	optional, the input description
outputs	map	optional, the output description
accept	function	optional, the accept function, make signal pass when returning true, block signal when returning false

function accept(signal : Signal) : boolean

If no ‘accept’ function specified, the filter accepts all signals

when ( comment, inputs, outputs, accept )

Create a filter

// only allow the signals, which have the 'age' greater than 18, pass
const filter = ns.when(signal => {
  return signal.get("age") > 18;
});

An alias of filter(). See filter() function for more detail

Signature

#NS.when(comment : string, inputs : map, outputs : map, accept : function) : Node

actuator ( comment, inputs, outputs, act )

Create an actuator

const ns = collar.ns('collarjs.demo.actuator');
// an http request actuator
const request = require('request');
const httpRequestActuator = ns.actuator("make http request",
  (signal, done) => {
    let url = signal.get("url");
    request(url, function (error, response, body) {
      done(error, body);  // put the page body to the outgoing signal's result
    })
  });

// print collarjs.com home page
ns.just({url:"http://collarjs.com"})  // create a single signal source
.to("make http request", httpRequestActuator) // to http request actuator
.do("print response body", signal => {  // print result
  console.log(signal.getResult());
})
.sink();  // drive the passive source

Make side effects. An actuator does not change the incoming signal (the result of the actuator will be added to a special field in signal, __result__, see Signal class for more details), it interacts with external worlds.

Asynchronous Operator (act function accept a callback)

Signature

#NS.actuator(comment : string, inputs : map, outputs : map, act : function) : Node

Parameters

Parameter	Type	Description
comment	string	optional, the comment of the filter
inputs	map	optional, the input description
outputs	map	optional, the output description
act(signal, done)	function	optional, the act function, it takes a callback function as the second argument, which accepts an error and result. If the error argument is not empty, an error signal will be sent, otherwise, the result will be injected to the signal as `__result__` field

function act(signal : Signal, done : function) : void

done argument is a callback function with signature:

function done(error : Error, result : any) : void

If error argument is not null, an ERROR signal is sent to stream.

If result argument is not null, the result is injected to the signal with a special name : __result__. You can access it with signal.getResult()

actuatorSync ( comment, inputs, outputs, actSync )

Create a sync actuator

const ns = collar.ns('collarjs.demo.actuatorSync')
ns
  .just({data : "Hello World"})  // create a single signal source
  .actuatorSync("print data", signal => {  // print signal data
    console.log(signal.get("data"));
  })
  .sink();  // drive the passive source

The synchronous version of actuator() api. The actSync function does not take a callback function as argument. Its return value will be injected to signal’s __result__ field.

Alias

do()

Signature

#NS.actuatorSync(comment : string, inputs : map, outputs : map, actSync : function) : Node

Parameters

Parameter	Type	Description
comment	string	optional, the comment of the filter
inputs	map	optional, the input description
outputs	map	optional, the output description
actSync(signal, done)	function	optional, the act function, it throws Error or returns the result. If an error is threw, an error signal will be sent, otherwise, the return value will be injected to the signal as `__result__` field

function actSync(signal : Signal) : any

If error is threw, an ERROR signal is sent to stream, otherwise the return value is injected to the signal with a special name : __result__. You can access it with signal.getResult()

do ( comment, inputs, outputs, actSync )

Create a sync actuator

collar
  .just({data : "Hello World"})  // create a single signal source
  .do("print data", signal => {  // print signal data
    console.log(signal.get("data"));
  })
  .sink();  // drive the passive source

An alias of actuatorSync(), see actuatorSync() for more details.

Signature

#NS.do(comment : string, inputs : map, outputs : map, actSync : function) : Node

processor ( comment, inputs, outputs, process )

Create a processor

const ns = collar.ns('collarjs.demo.processor')
const proc = ns.processor("double the payload",
  ( signal, done ) => {
    let payload = signal.payload;
    done(null, signal.new(signal.payload * 2));
  });

A processor modifies the incoming signal and emits it. It does not interact with external world. The process function takes a callback function to emit an error signal or the modified signal.

Signature

#NS.processor(comment : string, inputs : map, outputs : map, process : function) : Node

Parameters

Parameter	Type	Description
comment	string	optional, the comment of the filter
inputs	map	optional, the input description
outputs	map	optional, the output description
process(signal, done)	function	optional, the process function, it takes a callback function as the second argument, which accepts an error and new signal. If the error argument is not empty, an error signal will be sent, otherwise, the new signal will be sent

function process(signal : Signal, done : function) : void

done argument is a callback function with signature:

function done(error : Error, newSignal : Signal) : void

If error argument is not null, an ERROR signal is sent to stream otherwise the new signal is emitted.

processorSync ( comment, inputs, outputs, processSync )

Create a processor with processorSync

const ns = collar.ns('collarjs.demo.processorSync')
const proc = ns
  .processorSync("double the payload",
    ( signal ) => {
      return signal.new(signal.payload * 2);
    });

The synchronous version of processor(). The processSync function does not take a callback function as argument, instead, its returned value is sent as the outgoing signal.

Alias

map()

Signature

#NS.processorSync(comment : string, inputs : map, outputs : map, processSync : function) : Node

Parameters

Parameter	Type	Description
comment	string	optional, the comment of the filter
inputs	map	optional, the input description
outputs	map	optional, the output description
processSync(signal)	function	optional, the process function, it throws Error or returns the new signal. If an error is threw, an error signal will be sent, otherwise, the returned signal will be emitted

function processSync(signal : Signal) : Signal

If error is threw, an ERROR signal is sent to stream, otherwise the return signal is emitted

map ( comment, inputs, outputs, processSync )

Create a processor with processorSync

const ns = collar.ns('collarjs.demo.map');
const proc = ns
  .map("double the payload",
    ( signal ) => {
      return signal.new(signal.payload * 2);
    });

The alias of processorSync(). See processorSync() for more details.

Signature

#NS.map(comment : string, inputs : map, outputs : map, processSync : function) : Node

node ( comment, options )

To create a node, who doubles the odds input

// a node turn odds to an even
const ns = collar.ns('collarjs.demo.node');
var oddDoubler = ns.node({
  onSignal : function(signal){
    if (signal.payload % 2 != 0) {
      // only allow odds pass
      // do some side effect
      console.log(signal.payload);
      // change input signal
      const outSignal = signal.new(signal.payload * 2);
      // send it to next node
      this.send(outSignal);
    } else {
      // signal payload is even
      // don't make it pass, and request next signal
      this.request();
    }
  }
});

To create an error handler.

var errorHandler = ns.node({
  onError : function (signal, rethrow) {
    // print error
    console.error(signal.error);
    rethrow(signal);
  }
});

To create a sensor with 'watch’, which send timeout signal in 10 seconds

var evenPassFilter = ns.node({
  watch : function (options) {
    setTimer(() => {
      this.send({
        event : "timeout"
      });
    }, 10000);
  }
});

To create a filter with 'accept’.

var evenPassFilter = ns.node({
  accept : function (signal) {
    return signal.payload % 2 == 0;
  }
});

To create an actuator with 'act’, and put 'printed’ as result

var printPayloadActuator = ns.node({
  act : function (signal, done) {
    console.log(signal.payload);
    done(null, "printed")
  }
});

To create a processor with 'process’, and increase payload by 1

var addOneActuator = ns.node({
  process : function (signal, done) {
    done(null, signal.new(signal.payload + 1));
  }
});

It is not suggested to create node directly with this function. Use sensor, filter, actuator, or processor, unless your node needs multiple responsibilities. See “Single Responsibility Principle”

A general node constructor. It provides two sets of APIs for you to override in options:

high level API : watch, accept, act, process
low level API : onSignal, onError, onEnd

Signature

#NS.node(comment : string, options : map) : Node

Parameters

Parameter	Type	Description
comment	string	optional, the comment of the node
options	object	optional, the options, see Options section

Options

Option	Type	Description
onSignal(signal)	function	process incoming signal, you can call `this.send(signal)` to send the signal to next node, or call `this.request()` to ask for next signal.
onError(signal, rethrow)	function	process error signal, this function will be called if an `Error` signal enters your node. By default, this function blocks the signal. You can use `rethrow(signal)` function to rethrow the `Error` signal to the flow
onEnd(signal)	function	process END signal, this function will be called if and `END` signal enters your node.
watch(options)	function	watch the environment, send signal when necessary.
accept(signal)	function	accept a signal to process or not. Accept the signal if it returns true, otherwise block it.
act(signal, done)	function	make side effects and put the result in a special field (`__result__`) of the signal by calling `done(error, result)`. Do NOT change the signal payload in this function
process(signal, done)	function	modify the incoming signal, and send a new signal to next node by calling `done(error, newSignal)`

to ( comment, inputNode, outputNode, asActuator )

Connect to nodes

const ns = collar.ns('collarjs.demo.to');
var filter = ns.when("payload is odd",
  signal => signal.payload % 2 != 0)

var processor = ns.do("double payload",
  signal => signal.new(signal.payload * 2));

filter.to(processor);

Connect current node to another pipeline. This API returns the tail of the pipeline so that you can chain nodes in a fluent way.

Chain multiple nodes

var filter = ns.when("payload is odd",
  signal => signal.payload % 2 != 0)

var dbProcessor = ns.do("double payload",
  signal => signal.new(signal.payload * 2));

var incProcessor = ns.do("increase payload by 1",
  signal => signal.new(signal.payload + 1));

source
  .to(filter)
  .to(dbProcessor)
  .to(incProcessor);

Signature

#Node.to(comment : string, inputNode: Node, outputNode : Node, asActuator: boolean = false) : Node

Parameters

Parameter	Type	Description
comment	string	optional, the comment of the node
inputNode	Node	the first node of the pipeline to connect to
outputNode	Node	optional, the last node of the pipeline in order to chain to next node, default the same as inputNode
asActuator	boolean	optional, act as an actuator or as a processor (default: false)

ns ( namespace, tags )

To create a namespace with two perspectives : module and author

const testNS = collar.ns("com.collarjs.test", {module: "demo.test", author: "Bo"})

Make sure to replace com.collarjs.test with your namespace

// helloWorldActuator belongs to "com.collarjs.test" namespace
var helloWorldActuator = testNS.do(signal => {
  console.log("Hello World");
});

ns( namespace ) API is only available at top level collar object.

Create a new namespace. You can organize your flow with different namespaces. You can specify a tags parameter to show a different perspective of your code.

For example: module: 'demo.test'. You can see how your module interacts with each other in collar dev tool.

Signature

#collar.ns(namespace : string, tags: Map) : namespace

Parameters

Parameter	Type	Description
namespace	string	the namespace name
tags	Map	other perspectives

Return value

a new collar.js namespace

Utils API

just ( value )

Create a passive source with only one signal

collar.just(1)
  .do(signal => {
    console.log(signal.payload);
  })
  .sink();

// output
1

Create a passive source with only one signal. A sink is required to drive the passive source

Signature

#NS.just( value ) : Node

Parameters

Parameter	Type	Description
value	any	the signal data

asList ( values )

Create a passive source with a list of signal

collar.atList([1,2,3])
  .do(signal => {
    console.log(signal.payload);
  })
  .sink();

// output
1
2
3

Create a passive source with a list of signals. A sink is required to drive the passive source

Signature

#NS.asList( values ) : Node

Parameters

Parameter	Type	Description
values	array	the list of data

sink ()

drive a passive source

collar.just(1)
  .do(signal => {
    console.log(signal.payload);
  })
  .sink();

// output
1

Drive a passive source. A passive source is source who sends signal on demand. This is called back-pressure.

Signature

#NS.sink( ) : Node

errors ( comment, inputs, outputs, errorHandler )

Create a error handler node

const errorHandler = collar.errors((signal, rethrow) => {
  console.error(signal.error);
  rethrow(signal);
});

Handle error signal in the stream. By default, the error signal is blocked from propagating in the stream.

Signature

#NS.errors(comment : string, inputs : map, outputs : map, errorHandler : function) : Node

Parameters

Parameter	Type	Description
comment	string	optional, the comment of the sensor
inputs	map	optional, the input description
outputs	map	optional, the output description
errorHandler	function	the error handler function

function errorHandler(signal : Signal, rethrow : function) : void

Pass a errorHandler function as parameter to handle the error signal. The error could be obtained from signal’s error field. By default, the signal is blocked by the error handler, you can keep propagating the error signal by rethrow it using rethrow method.

Signal class

Signal is an envelop of stream data. It is immutable.

Properties

Get signal id

var s = new Signal({
  greeting : "hello world"
});

console.log(s.payload); // {greeting: "hello world"}

console.log(s.id);  

console.log(s.seq);

console.log(s.error); // the error, if the signal represents an error

console.log(s.end); // return true if signal represents an END signal

id

String. The id of the signal

seq

String. The alias of signal id

payload

Any. The payload of the signal

anonPayload

Any. The anonymous payload. When you create a signal with primitive type data, it will be store in anonymous payload, with a key __anon__;

error

Error. The error in the signal, if any

end

boolean. If the signal is an END signal

Constructor ( payload )

Create a signal

const Signal = collar.Signal;

var s = new Signal("payload");

Signature

Signal:constructor(payload)

Parameters

Parameter	Type	Description
payload	any	the payload of the signal

new ( payload )

Create a new signal from existing signal, keep the signal id

const Signal = collar.Signal;

var s1 = new Signal(1);
var s2 = s1.new(2);

console.log(s1.payload); // => 1
console.log(s2.payload); // => 2
console.log(s1.id == s2.id); // true

Create a new signal from existing signal and keep the signal id.

It is recommended to use this method to create a signal rather than creating a signal with constructor. This method keeps the signal id so that it is easy to track your data flow.

Signature

Signal:new(payload)

Parameters

Parameter	Type	Description
payload	any	the payload of the signal

get ( path )

get signal payload data

const Signal = collar.Signal;

var s = new Signal({
  greeting : "hello world"
  user : {
    address : {
      city : "Paris"
    }
  },
  films : [
    "Avengers",
    "X-man"
  ]
});

console.log(s.get("greeting")); // => hello world
console.log(s.get("user.address.city")); // => Paris
console.log(s.get(["user", "address", "city"])); // => Paris
console.log(s.get(["films.0"])); // => Avengers

Get the data in the signal by data path. You can use a string or an array to specify the json path.

Signature

Signal:get( path ) : any

Parameters

Parameter	Type	Description
path	string or array	the path of the data to access

set ( path, value )

set signal payload data at given json path

const Signal = collar.Signal;

var s = new Signal({});
s.set("user.address.city", "New York");
console.log(s.get(["user", "address", "city"])); // => New York

Set the data by json path. You can use a string or an array to specify the json path.

Signature

Signal:set( path, value ) : Signal

Parameters

Parameter	Type	Description
path	string or array	the path of the data to access
value	any	the data

del ( path )

delete signal payload data at given path

const Signal = collar.Signal;

var s = new Signal({
  greeting : "hello world"
  user : {
    address : {
      city : "Paris"
    }
  },
  films : [
    "Avengers",
    "X-man"
  ]
});

s.del("user.address");
console.log(s.get("user")) // => {}

Delete the data in the signal by data path. You can use a string or an array to specify the json path.

Signature

Signal:del( path ) : Signal

Parameters

Parameter	Type	Description
path	string or array	the path of the data to access

getResult ( )

get result of last actuator

collar.just(1)
  .do(signal => {
    return signal.payload * 2;
  })
  .do(signal => {
    console.log(signal.getResult()); // => 2
  })
  .sink();

Get the result of last actuator. Actuator will store the result to a special field __result__. This function is a shortcut of signal.get("__result__");

Signature

Signal:getResult() : any

setPayload ( payload )

set signal payload

const Signal = collar.Signal;

var s = new Signal(1);
console.log(s.anonPayload); // => 1
var s1 = s.setPayload(2);

// s' payload is always 1, as signal is immutable
console.log(s.anonPayload); // => 1
console.log(s1.anonPayload); // => 2
console.log(s.id == s1.id); // => true

Create a new signal with new payload, it is an alias of new()

Signature

Signal:setPayload( payload ) : Signal

Parameters

Parameter	Type	Description
value	any	the data

Dev client

Installation

Use npm to install collar.js dev client

npm install collar.js-dev-tool --save-dev

Use npm to install collar.js dev client for nodejs application. Use webpack to package it to your bundle for browser application. Or download the minified version for your browser from collarjs.com

Usage

require collar.js-dev-client at the beginning of your application

require("collar.js-dev-client");

// your collar flow here

or directly include it in your web page

<script src="js/collar.min.js"></script>
<script src="js/collar-dev-client.min.js"></script>
<script>
// your flow goes here
</script>

nodejs

require collar.js-dev-client at the beginning of your application.

browser

include collar-dev-client.min.js in your web page