README

About

FutoIn EventStream is fundamental part for efficient global state update in distributed systems. It is used for reliable event delivery, decentralized cache invalidation, efficient online segregation of active and warehouse data.

Unlike various message/event brokers, the focus of FutoIn Event Stream is integration with database transactions for reliable efficient event recording and delivery.

The design is not focused on high throughput as the primary reason is reliable association of events with database changes. Please consider using pure message brokers for throughput-intensive cases.

Documentation --> FutoIn Guide

Reference implementation of:

FTN18: FutoIn Interface - Event Stream
Version: 1.1

Spec: FTN18: FutoIn Interface - Event Stream v1.x

Features

Database transaction-bound event generation.
Standalone event generation.
Event polling for simple, but less efficient solutions.
Event pushing for advanced efficient cases.

Supported database types

MySQL
PostgreSQL
SQLite
Potentially, any other SQL-compliant supported by futoin-database

Installation for Node.js

Command line:

$ yarn add futoin-eventstream

$ npm install futoin-eventstream --save

Concept

Each event has auto-generated ID, type, data and timestamp. Type is all upper case identifier. Data is arbitrary JSON-friendly data.

Two configurable delivery strategies are supported: polling and streaming, but consumer acts as client in both cases.

There are two delivery modes: reliable and live. The later allow messages to be skipped. To ensure that events are reliably delivered, each consumer must register first.

Two message storage types are assumed: active small high performance area and slower data warehouse for all time history. DBEventArchiver tool is provided for efficient reliable data transfer.

More detailed concept is in the FTN18 spec.

Examples

1. Adding standalone events

GenFace.register(as, ccm, 'evtgen', endpoint );
// ...
const evtgen = ccm.iface('evtgen');
evtgen.addEvent(as, 'NULL_EVENT');
evtgen.addEvent(as, 'INT_EVENT', 123);
evtgen.addEvent(as, 'STR_EVENT', 'Some Str');
evtgen.addEvent(as, 'ARRAY_EVENT', [1, 2, 3]);
evtgen.addEvent(as, 'OBJECT_EVENT', { a: 1, b: 2, c: 3 });

2. Adding events in database transaction

For more advanced cases, you can check source code of DBGenFace#addXferEvent() to build more tailored statements.

DBGenFace.register(as, ccm, 'evtgen', endpoint );
// ...
const evtgen = ccm.iface('evtgen');
const db = ccm.db();

const xfer = db.newXfer();
xfer.insert('SomeTable').set('name', 'Name');
evtgen.addXferEvent(xfer, 'NEW_ENTRY', {name: 'Name'});
xfer.execute(as);

3. Poll for events with different components & filters

Each consumer is identifier by credentials + arbitrary component name.

As special "LIVE" component can be used for unreliable delivery.

PollFace.register(as, ccm, 'evtpoll', endpoint );
// ...
const evtpoll = ccm.iface('evtpoll');

// User info polling
evtpoll.registerConsumer(as, 'Security');
evtpoll.pollEvents(as, 'Security', last_known_id_here, ['USR_ADD', 'USR_MOD', 'USR_LOGIN']);
as.add((as, events) => {
    // ....
});

// Anti-Fraud processing
evtpoll.registerConsumer(as, 'AntiFraud');
evtpoll.pollEvents(as, 'AntiFraud', last_known_id_here, ['DEPOSIT', 'WITHDRAW', 'XFER']);
as.add((as, events) => {
    // ....
});

4. Event push with different components & filters

A child class of ReliableReceiverService should be created to properly handle incoming events.

A bi-directional channel like WebSockets or Internal must be used.

A separate Executor instance should be created for use in endpoint callbacks.

class UserReceiver extends ReliableReceiverService
{
    _onEvents( as, reqinfo, events ) {
        // ...
    }
}

const executor = Executor(ccm);
PollFace.register(as, ccm, 'evtpushsec', endpoint, credentials, { executor } );
UserReceiver.register(as, executor);

const evtpushsec = ccm.iface('evtpushsec');
evtpushsec.registerConsumer(as, 'Security');
evtpushsec.readyToReceive(as, 'Security', ['USR_ADD', 'USR_MOD', 'USR_LOGIN']);

5. Event history transfer

There should be a single system-wide instance of DBEventArchiver tool. The tool will automatically reconnect on errors. Processing state can be monitored through events.

DBAutoConfig(as, ccm, {
    evtdwh: {}
});
const archiver = new DBEventArchiver(ccm);

archiver.on('workerError', () => { ... });
archiver.on('receiverError', () => { ... });
archiver.on('newEvents', () => { ... });

// keep going until stopped
archiver.start(push_endpoint, credentials);

// to stop - automatically called on ccm.close()
archiver.stop();

6. Discarding event from active database

For performance and disaster recovery time reasons, operation critical database should be kept as small as possible. Events delivered to all consumers including DBEventArchiver can be removed the following way.

DBAutoConfig(as, ccm, {
    evt: {}
});
const discarder = new DBEventDiscarder(ccm);

archiver.on('workerError', () => { ... });
archiver.on('eventDiscard', () => { ... });

// keep going until stopped
discarder.start(ccm);

// to stop - automatically called on ccm.close()
discarder.stop();

7. Complete example

DBPushService inherits DBPollService, so there is no need to instance both.

This case show internal communicaton channel.

const ccm = new AdvancedCCM();
DBAutoConfig(as, ccm, {
    evt: {}
});
const executor = new Executor(ccm); // or NodeExecutor() for WebSockets
DBGenService.register(as, executor);
DBPushService.register(as, executor);

GenFace.register(as, ccm, 'evtgen', executor);
PollFace.register(as, ccm, 'evtpoll', executor);

const p = as.parallel();

p.loop( (as) => {
    ccm.iface('evtgen').addEvent(as, 'EVT', 'data');
});

p.add( (as) => {
    let last_id = null;
    
    as.loop( (as) => {
        ccm.iface('evtpoll').pollEvents(as, 'LIVE', last_id);
        
        as.add((as, events) => {
            if (events.length) {
                last_id = events[events.length - 1].id;
            } else {
                const timer = setTimeout( () => as.success(), 1e3 );
                as.setCancel((as) => clearTimeout(timer));
            }
        });
    });
});

API documentation

The concept is described in FutoIn specification: FTN18: FutoIn Interface - Event Stream v1.x

Classes

DBEventArchiver

Database Event Archiver service.

DBEventDiscarder

DB-specific event discarding.

It's assumed to be run against "active" database part as defined in the concept to reduce its size after all reliably delivered events are delivered to consumers.

Event are deleted based on limit_at_once to avoid too large transactions which may affect performance of realtime processes and break some DB clusters like Galera.

DBGenFace

GenFace for DB backend.

The only difference to original GenFace is native DB-specific API.

DBGenService

Database-specific event generation service

DBPollService

Database-based Poll Service

DBPushService

Database-specific Push Service

DBServiceApp

All-in-one DB EventStream initialization

GenFace

Event Stream - Generator Face

GenService

Event Stream - Generator Service Base

LiveReceiver

Reliable Event Receiver helper to minimize boilerplate code in projects.

PollFace

Event Stream - Poll Face

PollService

Event Stream - Poll Service Base

PushFace

Event Stream - Push Face

PushService

Event Stream - Push Service Base

ReceiverFace

Event Stream - Receiver Face

ReceiverService

Base implementation for receiver side

ReliableReceiver

Reliable Event Receiver helper to minimize boilerplate code in projects.

ReliableReceiverService

Base implementation for reliable receiver side.

DBEventArchiver

Database Event Archiver service.

Kind: global class
Note: No more than one instance should run at once.

new DBEventArchiver(db_ccm)

C-tor

Param	Type	Description
db_ccm	`AdvancedCCM`	CCM instance with registered '#db.evtdwh' interface

DBEventDiscarder

DB-specific event discarding.

It's assumed to be run against "active" database part as defined in the concept to reduce its size after all reliably delivered events are delivered to consumers.

Event are deleted based on limit_at_once to avoid too large transactions which may affect performance of realtime processes and break some DB clusters like Galera.

Kind: global class

DBEventDiscarder

dbEventDiscarder.start(ccm, [options])

Start event discarding

Kind: instance method of DBEventDiscarder

Param	Type	Default	Description
ccm	`AdvancedCCM`		CCM with registered #db.evt interface
[options]	`object`	`{}`	options
[options.poll_period_ms]	`integer`	`600e3`	poll interval
[options.limit_at_once]	`integer`	`1000`	events to delete at once
[options.event_table]	`string`	`"default"`	events table
[options.consumer_table]	`string`	`"default"`	consumers table

dbEventDiscarder.stop()

Stop event discarding

Kind: instance method of DBEventDiscarder

"workerError"

Emitted on worker errors

Kind: event emitted by DBEventDiscarder

"eventDiscard"

Emitted on discarded events

Kind: event emitted by DBEventDiscarder

DBGenFace

GenFace for DB backend.

The only difference to original GenFace is native DB-specific API.

Kind: global class

DBGenFace
- .DB_EVENT_TABLE ⇒ string
- .addXferEvent(xb, type, data, [table])

dbGenFace.DB_EVENT_TABLE ⇒ `string`

Easy access to DB event table name

Kind: instance property of DBGenFace
Returns: string - raw table name

dbGenFace.addXferEvent(xb, type, data, [table])

Helper to add event generation into DB transaction

Kind: instance method of DBGenFace

Param	Type	Default	Description
xb	`XferBuilder`		instance of transaction builder
type	`string`		event type
data	`*`		any data
[table]	`string`	`"evt_queue"`	event queue

DBGenService

Database-specific event generation service

Kind: global class

new DBGenService(_as, executor, [options])

Please use DBGenService.regster()

Param	Type	Default	Description
_as	`AsyncSteps`		async step interface
executor	`Executor`		related Executor
[options]	`object`	`{}`	options
[options.event_table]	`string`	`"default"`	events table

DBPollService

Database-based Poll Service

Kind: global class

new DBPollService(as, executor, [options])

Please use DBPollService,register()

Param	Type	Default	Description
as	`AsyncSteps`		async step interface
executor	`Executor`		related Executor
[options]	`object`	`{}`	options
[options.event_table]	`string`	`"default"`	events table
[options.consumer_table]	`string`	`"default"`	consumers table

DBPushService

Database-specific Push Service

Kind: global class

new DBPushService(as, executor, [options])

Please use DBPushService,register()

Param	Type	Default	Description
as	`AsyncSteps`		async step interface
executor	`Executor`		related Executor
[options]	`object`	`{}`	options
[options.event_table]	`string`	`"default"`	events table
[options.consumer_table]	`string`	`"default"`	consumers table
[options.sleep_min]	`integer`	`100`	minimal sleep on lack of events
[options.sleep_max]	`integer`	`3000`	maximal sleep on lack of events
[options.sleep_step]	`integer`	`100`	sleep time increase on lack of events

DBServiceApp

All-in-one DB EventStream initialization

Kind: global class

DBServiceApp
- new DBServiceApp(as, options)
- .ccm() ⇒ AdvancedCCM
- .executor() ⇒ Executor
- .close([done])

new DBServiceApp(as, options)

C-tor

Param	Type	Default	Description
as	`AsyncSteps`		AsyncSteps interface
options	`object`	`{}`	options
[options.ccm]	`AdvancedCCM`		external CCM instance
[options.executor]	`Executor`		external private executor instance
[options.ccmOptions]	`object`		auto-CCM options
[options.notExpectedHandler]	`callable`		'notExpected' error handler
[options.executorOptions]	`object`		private auto-Executor options
[options.evtOptions]	`object`		eventstream options
[options.discarderOptions]	`object`		discarder options
[options.enableDiscarder]	`boolean`		enable discarder, if true
[options.archiverOptions]	`object`		discarder options
[options.enableArchiver]	`boolean`		enable archiver, if true

dbServiceApp.ccm() ⇒ `AdvancedCCM`

CCM instance accessor

Kind: instance method of DBServiceApp
Returns: AdvancedCCM - instance

dbServiceApp.executor() ⇒ `Executor`

Executor instance accessor

Kind: instance method of DBServiceApp
Returns: Executor - instance

dbServiceApp.close([done])

Shutdown of app and related instances

Kind: instance method of DBServiceApp

Param	Type	Default	Description
[done]	`callable`		done callback

GenFace

Event Stream - Generator Face

Kind: global class

GenFace

GenFace.LATEST_VERSION

Latest supported FTN17 version

Kind: static property of GenFace

GenFace.PING_VERSION

Latest supported FTN4 version

Kind: static property of GenFace

GenFace.register(as, ccm, name, endpoint, [credentials], [options])

CCM registration helper

Kind: static method of GenFace

Param	Type	Default	Description
as	`AsyncSteps`		steps interface
ccm	`AdvancedCCM`		CCM instance
name	`string`		CCM registration name
endpoint	`*`		see AdvancedCCM#register
[credentials]	`*`		see AdvancedCCM#register
[options]	`object`	`{}`	interface options
[options.version]	`string`	`"1.0"`	interface version to use

GenService

Event Stream - Generator Service Base

Kind: global class

GenService.register(as, executor, options) ⇒ `GenService`

Kind: static method of GenService
Returns: GenService - instance

Param	Type	Description
as	`AsyncSteps`	steps interface
executor	`Executor`	executor instance
options	`object`	implementation defined options

LiveReceiver

Reliable Event Receiver helper to minimize boilerplate code in projects.

Kind: global class

LiveReceiver

new LiveReceiver(executor_ccm)

Initialize event archiver.

Param	Type	Description
executor_ccm	`AdvancedCCM`	CCM for executor

liveReceiver.start(endpoint, [credentials], [options])

Start receiving events for archiving

Kind: instance method of LiveReceiver
Note: options.executor is overridden

Param	Type	Default	Description
endpoint	`*`		see PushFace
[credentials]	`*`		see PushFace
[options]	`*`	`{}`	see PushFace
[options.component]	`string`		component name
[options.want]	`array`		"want" parameter for event filtering

liveReceiver.stop()

Stop receiving events

Kind: instance method of LiveReceiver

liveReceiver._registerReceiver(as, executor, options) ⇒ `ReceiverService`

Override to register custom instance of ReceiverService.

Kind: instance method of LiveReceiver
Returns: ReceiverService - instance of service

Param	Type	Description
as	`AsyncSteps`	async steps interface
executor	`Executor`	Internal Executor instance
options	`object`	passed options

liveReceiver._onEvents(as, events)

Override to catch new events here instead of using newEvents event handler.

Kind: instance method of LiveReceiver

Param	Type	Description
as	`AsyncSteps`	async steps interface
events	`array`	array of events

"receiverError"

Emitted on not expected receiver errors

Kind: event emitted by LiveReceiver

"workerError"

Emitted on worker errors

Kind: event emitted by LiveReceiver

"newEvents"

Emitted on new events

Kind: event emitted by LiveReceiver

"ready"

Emitted after event receiver is ready

Kind: event emitted by LiveReceiver

PollFace

Event Stream - Poll Face

Kind: global class

PollFace

PollFace.LATEST_VERSION

Latest supported FTN17 version

Kind: static property of PollFace

PollFace.PING_VERSION

Latest supported FTN4 version

Kind: static property of PollFace

PollFace.register(as, ccm, name, endpoint, [credentials], [options])

CCM registration helper

Kind: static method of PollFace

Param	Type	Default	Description
as	`AsyncSteps`		steps interface
ccm	`AdvancedCCM`		CCM instance
name	`string`		CCM registration name
endpoint	`*`		see AdvancedCCM#register
[credentials]	`*`		see AdvancedCCM#register
[options]	`object`	`{}`	interface options
[options.version]	`string`	`"1.0"`	interface version to use

PollService

Event Stream - Poll Service Base

Kind: global class

PollService.register(as, executor, options) ⇒ `PollService`

Kind: static method of PollService
Returns: PollService - instance
Note: Chunk event count is lower then protocol permits by default as there is a typical amount 64K futoin message limit.

Param	Type	Default	Description
as	`AsyncSteps`		steps interface
executor	`Executor`		executor instance
options	`object`		implementation defined options
[options.allow_reliable]	`boolean`	`true`	allow reliable consumers
[options.allow_polling]	`boolean`	`true`	allow polling calls
[options.max_chunk_events]	`integer`	`100`	maxium events per request

PushFace

Event Stream - Push Face

Kind: global class

PushFace.register(as, ccm, name, endpoint, [credentials], [options])

CCM registration helper

Kind: static method of PushFace

Param	Type	Default	Description
as	`AsyncSteps`		steps interface
ccm	`AdvancedCCM`		CCM instance
name	`string`		CCM registration name
endpoint	`*`		see AdvancedCCM#register
[credentials]	`*`		see AdvancedCCM#register
[options]	`object`	`{}`	interface options
[options.version]	`string`	`"1.0"`	interface version to use

PushService

Event Stream - Push Service Base

Kind: global class

PushService
- instance
  - "pushError"
  - "queueOverflow"
- static
  - .register(as, executor, options) ⇒ PushService

"pushError"

Emitted in push error handlers

Kind: event emitted by PushService

"queueOverflow"

Emitted in push error handlers

Kind: event emitted by PushService

PushService.register(as, executor, options) ⇒ `PushService`

Kind: static method of PushService
Returns: PushService - instance

Param	Type	Default	Description
as	`AsyncSteps`		steps interface
executor	`Executor`		executor instance
options	`object`		implementation defined options
[options.allow_reliable]	`boolean`	`true`	allow reliable consumers
[options.allow_polling]	`boolean`	`true`	allow polling calls

ReceiverFace

Event Stream - Receiver Face

Kind: global class

ReceiverFace
- .LATEST_VERSION
- .register(as, channel, [options]) ⇒ string

ReceiverFace.LATEST_VERSION

Latest supported FTN17 version

Kind: static property of ReceiverFace

ReceiverFace.register(as, channel, [options]) ⇒ `string`

CCM registration helper

Kind: static method of ReceiverFace
Returns: string - - iface:ver of registered interface

Param	Type	Default	Description
as	`AsyncSteps`		steps interface
channel	`ChannelContext`		Bi-Direction channel instance
[options]	`object`	`{}`	interface options
[options.version]	`string`	`"1.0"`	interface version to use

ReceiverService

Base implementation for receiver side

Kind: global class

ReceiverService
- instance
  - ._onEvents(as, reqinfo, events)
- static
  - .register(as, executor, options) ⇒ PushService

receiverService._onEvents(as, reqinfo, events)

Member to override to handle vents.

Kind: instance method of ReceiverService

Param	Type	Description
as	`AsyncSteps`	AsyncSteps interface
reqinfo	`RequestInfo`	request info object
events	`array`	list of events

ReceiverService.register(as, executor, options) ⇒ `PushService`

Kind: static method of ReceiverService
Returns: PushService - instance

Param	Type	Description
as	`AsyncSteps`	steps interface
executor	`Executor`	executor instance
options	`object`	implementation defined options

ReliableReceiver

Reliable Event Receiver helper to minimize boilerplate code in projects.

Kind: global class

ReliableReceiver
- ._registerReceiver(as, executor, options) ⇒ ReliableReceiverService
- "processedEvents"

reliableReceiver._registerReceiver(as, executor, options) ⇒ `ReliableReceiverService`

Override to register custom instance of ReliableReceiverService.

Kind: instance method of ReliableReceiver
Returns: ReliableReceiverService - instance of service

Param	Type	Description
as	`AsyncSteps`	async steps interface
executor	`Executor`	Internal Executor instance
options	`object`	passed options

"processedEvents"

Emitted for count of archived events in each iteration.

Kind: event emitted by ReliableReceiver

ReliableReceiverService

Base implementation for reliable receiver side.

Kind: global class
Note: Unlike ReceiverService, it restores proper order of events.

documented by jsdoc-to-markdown.

Usage no npm install needed!

README

About

Features

Supported database types

Installation for Node.js

Concept

Examples

1. Adding standalone events

2. Adding events in database transaction

3. Poll for events with different components & filters

4. Event push with different components & filters

5. Event history transfer

6. Discarding event from active database

7. Complete example

API documentation

Classes

DBEventArchiver

new DBEventArchiver(db_ccm)

DBEventDiscarder

dbEventDiscarder.start(ccm, [options])

dbEventDiscarder.stop()

"workerError"

"eventDiscard"

DBGenFace

dbGenFace.DB_EVENT_TABLE ⇒ string

dbGenFace.addXferEvent(xb, type, data, [table])

DBGenService

new DBGenService(_as, executor, [options])

DBPollService

new DBPollService(as, executor, [options])

DBPushService

new DBPushService(as, executor, [options])

DBServiceApp

new DBServiceApp(as, options)

dbServiceApp.ccm() ⇒ AdvancedCCM

dbServiceApp.executor() ⇒ Executor

dbServiceApp.close([done])

GenFace

GenFace.LATEST_VERSION

GenFace.PING_VERSION

GenFace.register(as, ccm, name, endpoint, [credentials], [options])

GenService

GenService.register(as, executor, options) ⇒ GenService

LiveReceiver

new LiveReceiver(executor_ccm)

liveReceiver.start(endpoint, [credentials], [options])

liveReceiver.stop()

liveReceiver._registerReceiver(as, executor, options) ⇒ ReceiverService

liveReceiver._onEvents(as, events)

"receiverError"

"workerError"

"newEvents"

"ready"

PollFace

PollFace.LATEST_VERSION

PollFace.PING_VERSION

PollFace.register(as, ccm, name, endpoint, [credentials], [options])

PollService

PollService.register(as, executor, options) ⇒ PollService

PushFace

PushFace.register(as, ccm, name, endpoint, [credentials], [options])

PushService

"pushError"

"queueOverflow"

PushService.register(as, executor, options) ⇒ PushService

ReceiverFace

ReceiverFace.LATEST_VERSION

ReceiverFace.register(as, channel, [options]) ⇒ string

ReceiverService

receiverService._onEvents(as, reqinfo, events)

ReceiverService.register(as, executor, options) ⇒ PushService

ReliableReceiver

reliableReceiver._registerReceiver(as, executor, options) ⇒ ReliableReceiverService

"processedEvents"

ReliableReceiverService

dbGenFace.DB_EVENT_TABLE ⇒ `string`

dbServiceApp.ccm() ⇒ `AdvancedCCM`

dbServiceApp.executor() ⇒ `Executor`

GenService.register(as, executor, options) ⇒ `GenService`

liveReceiver._registerReceiver(as, executor, options) ⇒ `ReceiverService`

PollService.register(as, executor, options) ⇒ `PollService`

PushService.register(as, executor, options) ⇒ `PushService`

ReceiverFace.register(as, channel, [options]) ⇒ `string`

ReceiverService.register(as, executor, options) ⇒ `PushService`

reliableReceiver._registerReceiver(as, executor, options) ⇒ `ReliableReceiverService`