oja-flow

This is the original Oja v1.0 that is now a subset of Oja v2.0+.

The module provides a way to assemble independent actions/events into a single flow based pub/sub pattern with event backlog, timeout support. It can map events to promises and back as well as chain them to other topics. Use it to structure your application business logic.

Idea

The main reason for creation of this module is to allow decoupling business logic into smaller isolated standalone components in the application via pub/sub API that can be used to compose higher async structures and easily tested in total isolation.

Architecture

Pub/sub flow

This approach is based on pub/sub interface. It accumulates events in the backlog for new subscribers. This has pros and cons:

• Allows consumers/producers to be added in any order and guarantees delivery of an event.
• Accumulates events in memory, thus it cannot be used for long running flows as it eventually will run out of memory.
• It is good fit for short request/stream flows that eventually complete and garbage collected.

Features:

• Queuing events for later consuption. For example, if consumer added after the event happened, it would still receive it.
• Consume events as stream/promises.
• Consume stream as events/promises
• Consume promises as stream/events
• Merge promise/streams/events into events/stream/pomises.
• Query the state of the whole flow to see what is still pending and queues states for each topic.
• Set resolution timeout for every topic, upon which it will provide what topic caused the timeout and pending topics.
• Compose/link flows into one.
• Debug/listen to all events going on in the flow.

Install

npm install @ebay/oja-flow --save

API

Flow API

The flow is generic definition of the control flow.

• Flow([baseFlow]) creates a flow object

  • baseFlow is an optional base flow that would be joined with a new flow.

• consume(topics[, callback]) adds a consumer to the flow for the given topic or topics

  • topics is one or more topics to capture by the handler
  • callback((data|map), flow) is a handler, if provided, it will be called once all topics are resolved
    • data resolved for the single topic
    • map of resolved data mapped to corresponding topics
    • flow is an instance of the flow for convenience
  • returns promise or a list of promises if callback is not provided;
    • promise for single topic
    • promises mapped to the provided topics

• consumeStream(topic[, callback]) returns a readable stream of events for the given topic

  • Note: if callback is provided, it will return a stream to the callback(stream) and continue cascading flow by returning 'this' reference;

• getReader(topic) returns a reader for a series of events published under given topic.

  • reader.next() returns a promise for a single event.
    • The reader reached the end of stream whenever the promise resolves to undefined

• define(topics[, data|promise|function]) defines a producer for the given topic or an array of topics

  • topics is a single topic or an array of topics to broadcast the data with.
  • data will be immediately published under the given topics into the flow; in case an error object is passed, the flow will be stopped and an 'error' event will be broadcasted into the flow.
  • promise will publish data once resolved; in case of reject, the flow will be stopped and an 'error' event will be broadcasted into the flow.
  • function(publisher, flow) will be called immediately
    • publisher is a convenience object to publish events for the assigned topic in async use-case
    • flow is an instance of the current flow.
    • if function returns a non-undefined value, it will be published under the assigned topic immediately; this is useful in a sync flow.

• on(topic, callback) adds a listener for the given topic.

• once(topic, callback) adds a listener for the given topic that will be called for the fist event.

• setMaxListeners(number) allows to increase a limit on how many listeners can be added to the given action, similar to event emitters.

• catch(callback) adds an error handler to the flow. If no handlers are registered, it will re-throw the error.

• timeout(topics, ms) sets a timeout for the topic or and array of topics. The timeout error will be generated if the topic(s) are not resolved within the specified period of time (in milliseconds)

• state() returns a list of topics and their state (topics queue, pending topics)

• Publisher API is a convenience object to publish events for the assigned topic

  • pub(data|promise)
    • data will be immediately published under the given topics into the flow; in case error object is passed, the flow will be stopped and 'error' event will be broadcasted into the flow.
    • promise will publish data once resolved; in case of reject the flow will be stopped and 'error' generated.

Action flow API

Action type is more specific to business unit that extends Flow further to allow defining executable independent business logic blocks called actions.

While flow starts executing immediately when one defines it, action is activate by calling execute method.

The actions can be composed into more complex actions. Once the action is activated, it will implicitly activate all children actions.

Actions cannot be added after they have been started.

• Action() is an action constructor

• add(action) adds actions to the main action

  • action can be a single action or an array of actions. Action parameter can also be a generic function that accepts flow reference:

• execute() is a method called by framework during activation; the action logic should be put into this method

• activate() starts the action and all its children. This method better not be overridden or one needs to make sure a base function called.

Usage

First we would like you to focus more on how you can apply this module to simplify your business logic with the use of Action type then more generic examples on how generic Flow type can be used and applied to the action type as it extends Flow.

Action

Execute an action

const Action = require('@ebay/oja-flow/action');
new Action().activate();

Execute an action that generates an event

const Action = require('@ebay/oja-flow/action');
class MyAction extends Action {
    execute() {
        this.define('foo', 'bar');
    }
}
new MyAction()
    .activate()
    .consume('foo', data => console.log(data)); // will print bar

Composing more complex actions out of basic ones

const Action = require('@ebay/oja-flow/action');
class Greet extends Action {
    execute() {
        this.define('greet', 'Hello');
    }
}
// demonstrate generic function instead of action object
async function who(flow) {
    // demonstrate re-mapping
    const name = await flow.consume('name');
    flow.define('who', name);
}
class Greeting extends Action {
    async execute() {
        const ['greet', 'who'] = await this.consume(['greet', 'who']);
        this.define('greeting', `${data.greet} ${data.who}`)
    }
}
const helloAction = new Greeting();
helloAction
    .add(new Hello())
    .add(who)
    .activate()
    .define('name', 'World')
    .consume('greeting', console.log); // prints Hello World

Flow

The usage examples are generic and look more like generic event pub/sub mode.

It can be used to create more specific controls like Action mentioned above.

Simple pub/sub

const Flow = require('@ebay/oja-flow/flow');
const flow = new Flow();

// create consumer component
const consumer = flow.consume('foo');
consumer.then(foo => {
    console.log(foo); // prints 'bar'
});

// define producer component
const producer = flow.define('foo');
// publish
producer.pub('bar');

Shorter form for clarity:

// create consumer component
flow
.consume('foo', foo => {
    console.log(foo); // prints 'bar'
})
.define('foo', 'bar');

Consuming multiple events for the given topic

// create consumer component
flow
.consume('foo', foo => {
    console.log(foo); // prints 'bar1' and 'bar2'
})
// generate events
.define('foo', 'bar1')
.define('foo', 'bar2');

Consuming events as a stream

const buffer = [];
// create consumer stream
const stream = flow.consumeStream('foo');
stream.on('data', data => buffer.push(data));
stream.on('end', () => {
    console.log(buffer); // prints one, two, three
})
// generate some data
flow.define('foo', 'one');
flow.define('foo', 'two');
flow.define('foo', 'three');
flow.define('foo', null);

Consuming events via reader

const Flow = require('.').Flow;
const buffer = [];
const flow = new Flow();
// create consumer stream
const reader = flow.getReader('foo');
// generate some data
flow.define('foo', 'one');
flow.define('foo', 'two');
flow.define('foo', 'three');
flow.define('foo', null);

async function read() {
    while(true) {
        const data = await reader.next()
        if (data === undefined) {
            break;
        }
        console.log(data);
    }
}

read();

Consuming multiple topics in one short

// consume multiple topics
flow.consume(['foo', 'qoo'], input => {
    console.log(input.foo);     // prints faa
    console.log(input.qoo);     // prints qaa
});
flow.define('foo', 'faa');
flow.define('qoo', 'qaa');

Using promise

// create consumer component
flow
.consume('foo', foo => {
    console.log(foo); // prints 'bar'
})
.define('foo', new Promise(resolve => {
    resolve('bar');
}));

Multiple consumers, single producer

// create consumer component
flow
.consume('foo', foo => {
    console.log(foo); // prints 'bar'
})
.consume('foo', foo => {
    console.log(foo); // prints 'bar'
})
.define('foo', 'bar');

Chaining actions, mixing, etc.

// NOTE: the order of consume/define does not matter
// create consumer component
flow
.consume('foo', (foo, runtime) => {
    console.log(foo); // prints 'faa'
    runtime.define('qoo', 'qaa'); // can consume and produce new data
})
.consume('qoo', (qoo, runtime) => {
    console.log(qoo); // prints 'qaa'
    runtime.define('woo', Promise.resolve('waa')); // can use async promise
})
// start chain reaction here
.define('foo', 'faa')   
// lets produce multiple events via event emitter
.consume('woo', (woo, runtime) => {    
    console.log(woo); // prints waa
    // define as event emitter
    const roo = runtime.define('roo');
    // simulate async flow with two event emitted
    setImmediate(() => {
        // generate multiple events
        roo.pub('raa1');
        roo.pub('raa2');
    });
})
// validate
.consume('roo', roo => {
    console.log(roo);   // prints raa1 and raa2
})
// consume multiple topics
.consume(['foo', 'qoo'], input => {
    console.log(input.foo);     // prints faa
    console.log(input.qoo);     // prints qaa
})
// can consume inside consume
.consume('foo', (foo, runtime) => {
    console.log(foo);     // prints faa
    runtime.consume('qoo', qoo => {
        console.log(input.qoo);     // prints qaa
    });
    // or
    flow.consume('qoo', qoo => {
        console.log(input.qoo);     // prints qaa
    });
})
// can generate multiple events using pub
.define('doo', publisher => {
    publisher.pub('daa1');
    publisher.pub('daa2');
    publisher.pub('daa3');
    publisher.pub(null);
})
.consume('doo', doo => {
    console.log(doo); // prints daa1, daa2, daa3, null
});
.consumeStream('doo', stream => {
    stream.on('data', console.log) // prints daa1, daa2, daa3
    stream.on('end', () => console.log('end of "doo"'));    
})
// NOTE: we can consume first event via promise if we are not interested in the rest
.consume('doo').then(doo => {
    console.log(doo); // prints daa1
});

// for debug you can listen to all events
flow.consume('*', evt => {
    console.log(`Event type: ${evt.name}, data: ${evt.data}`);
})

Join flows together

const base = new Flow();
base.define('foo', 'bar')
const flow = new Flow(base);
flow.consume('foo', foo => {
    console.log(foo); // prints bar
});

You can also make them depend on each other

const base = new Flow();
base.consume('shared', (_, rt) => {
    rt.define('foo', 'bar');
});
const flow = new Flow(base);
flow.consume('foo', foo => {
    console.log(foo); // prints bar
});
flow.define('shared', ''); // trigger the chain

Timeouts

The promise chain may be hard to figure out where it is blocked. Oja allows to set a timeout for the given topics and upon the timeout would provide an error message listing topics that have not been resolved yet.

flow
.consume('foo')
.consume('bar')
.consume('too')
.timeout(['foo', 'bar'], 300)   // 300 ms
.define('bar', 'boo')
.catch(err => {
    console.log(err.message); // prints "Topic/s (foo) timed out, pending topics (too)"
});

Querying for the state

Oja provides a current status of topcis that have not been resolved

flow
.consume('foo')
.consume('bar')
.consume('too')
.define('bar', 'boo');

console.log(flow.state()); // prints [foo, too]

Error

Throwing error

flow.define('error', new Error('Boom'));
// or
flow.define('error', Promise.reject(new Error('Boom')));
// or
flow.define('data', () => {
    return new Error('Boom');
});
// or
flow.define('data', runtime => {
    runtime.pub(new Error('Boom'));
});

Catching error

flow.catch(err => {
    console.log(err);   // prints Boom if linked to the above flow
});
// Or
flow.consume('error', err => {
    console.log(err);   // prints Boom if linked to the above flow
});

Error stops flow

The error will prevent further events including error events from publishing.

flow
.define('foo', 'faa')
.define('boo', 'baa')
.define('error', new Error('Boom'))
.define('too', 'taa')
.consume('foo', foo => {
    console.log(foo); // prints faa
})
.consume('boo', foo => {
    console.log(foo); // prints baa
})
.consume('too', too => {
    // will never happen
    throw new Error('Should never happen');
})
.catch(err => { // catch error
    console.log(err);   // print Boom
});

Composing complex flows

const base = new Flow();
new Foo(base)

Pipe

Declarative approach to build asynchronous pipelines.

const Pipe = require('oja/pipe');

const action = new Pipe()
    .next(data => {
        // transform data into array
        return data.split(',');
    })
    // map sync to async stream
    .map((itm, index) => {
        // this should reverse the order
        return new Promise(resolve => setTimeout(() => itm, (10 - index) * 100)
    })
    .filter((itm, index) => {
        // this is still streaming data
        // filtering only the ones in the below array
        return 'one,two,six'.indexOf('itm') !== -1;
    })
    .merge(items => {
        // aggregating streaming data
        console.log('merged items', items); // >> ["six", "two", "one"]
        return items;
    })
    .next(items => {
        // demonstrate async map
        return Promise.all(items)
    })
    .reduce((memo, itm) => {
        memo[itm] = itm;
        return memo;
    })
    .build();

const ret = await action('zero,one,two,three,four,five,six,seven,eight');
console.log(ret); // >>  { "six": "six", "two": "two", "one": "one"  }