Build a flow of components with Flow.load() and run it with exec()

Total.js 5 · total5@0.0.19 flowflowstreamcomponentsautomation Executed against Total.js 5

A FlowStream is a small program made of components (boxes with inputs and outputs) and connections between them. Total.js runs it inside your application: a message enters one component, the component sends results to its outputs, and the connections carry them to the next components. The same schema is what the Flow designer of Total.js draws; here it is written by hand.

Flow.load(schema, [callback]) starts a FlowStream and gives the callback (err, instance). The schema is an object:

id and name: the id is the key in Flow.instances (Flow.instances.demo); without an id a random one is made.
components: an object with the source code of each component, as text. The text is a script, <script total>...</script> (the form that the designer saves).
design: the instances of the components. Each one has an id, the component that it uses, its config and connections, an object whose keys are output ids and whose values are lists of { id, index }: the instance and the input id that receive what is sent to that output.
variables: values that components can read.

A component is a module: exports.id, exports.name, exports.inputs and exports.outputs (lists of { id, name }), exports.config (the defaults) and exports.make = function(instance, config). In make() the component sets instance.message = function($) {}, which receives each message: $.data is the data, $.send(outputid, data) sends it to an output and $.destroy() ends a message that does not go any further.

To use the flow from your code, instance.exec({ id, data, callback }) sends data to the instance id and calls callback(err, message) when the message ends; the last message of the chain is message (so message.data is what the last component received). instance.reconfigure(id, config) changes the configuration of a running instance and Flow.remove(id) stops the FlowStream and forgets it. FlowStream is the same object as Flow: two names for the manager.

A component is code that runs with all the rights of the application (in the same thread unless the schema says worker). Never load a schema or a component that a visitor can write.

definitions/flow.js

const upper = `<script total>
exports.id = 'upper';
exports.name = 'Upper case';
exports.inputs = [{ id: 'input', name: 'Input' }];
exports.outputs = [{ id: 'output', name: 'Output' }];
exports.config = { prefix: '' };
exports.make = function(instance, config) {
	instance.message = function($) {
		$.send('output', config.prefix + String($.data).toUpperCase());
	};
};
</script>`;

const collect = `<script total>
exports.id = 'collect';
exports.name = 'Collect';
exports.inputs = [{ id: 'input', name: 'Input' }];
exports.make = function(instance) {
	instance.message = function($) {
		MAIN.collected.push($.data);
		$.destroy();
	};
};
</script>`;

MAIN.collected = [];

MAIN.flow = null;

Flow.load({
	id: 'demo',
	name: 'Demo',
	components: { upper: upper, collect: collect },
	design: {
		a: { id: 'a', component: 'upper', config: { prefix: '> ' }, connections: { output: [{ id: 'b', index: 'input' }] } },
		b: { id: 'b', component: 'collect', config: {}, connections: {} }
	},
	variables: {}
}, function(err, instance) {
	MAIN.flow = instance;
});

controllers/flow.js

exports.install = function() {
	ROUTE('GET /shout/', shout);
	ROUTE('POST /prefix/ @json', prefix);
	ROUTE('GET /collected/', collected);
	ROUTE('GET /info/', info);
	ROUTE('POST /stop/', stop);
};

function shout($) {
	if (!MAIN.flow)
		return $.invalid('The flow is not running');
	MAIN.flow.exec({ id: 'a', data: $.query.text, callback: (err, message) => $.json({ result: message ? message.data : null }) });
}

function prefix($) {
	MAIN.flow.reconfigure('a', { prefix: $.body.prefix });
	$.json({ ok: true });
}

function collected($) {
	$.json(MAIN.collected);
}

function info($) {
	$.json({ flows: Object.keys(Flow.instances), instances: Object.keys(MAIN.flow.flow.meta.flow), same: Flow === FlowStream });
}

function stop($) {
	Flow.remove('demo');
	MAIN.flow = null;
	$.json({ flows: Object.keys(Flow.instances) });
}