Lightweight pub/sub module with event backlog, timeout support that maps events to promises and back as well as chains them to other topics. Use it to structure 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 and/or common context that leads to a simpler/isolated unit testing with easy mocking of input data and easy composition of components into higher async structures.
The context based approach allows a developer to slice the business logic into small, isolated busness actions that communicate to each other via context and that encourages developers to use the same action interface across application code, which makes the code more predictable, easy to define and test with mock data and yes, boring.
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 sonsumer 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.
Context/business actions
The module provides a way to slice the code into isolated, small and independent actions that can later be used to compose much more complex applications with dependency injection via context object.
Install
npm install oja -S
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
- The reader reached the end of stream whenever the promise resolves to
- reader.next() returns a promise for a single event.
-
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.
- pub(data|promise)
Action 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.
Context API
While pub/sub is based on knowledge of topics consumed and published, context based approach provides a more explicit way of what kind actions are available in application.
- createContext(options) - creates a context with actions and properties injected as part of options. It returns a context reference; which provides access to all other actions. It is also a Flow object, so one can mix flow with context. It is recommended to refrain from creating too many Flow/Actions objects and stick to the use of only one - context object - as it is passed everywhere anyways.
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 you will see Context based examples and 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 = Action;;
Execute an action that generates an event
const Action = Action; { this; } ; // will print bar
Composing more complex actions out of basic ones
const Action = Action; { this; }// demonstrate generic function instead of action object { // demonstrate re-mapping flow;} { this; }const helloAction = ;helloAction ; // prints Hello World
Context Action
Action definition
modulesexports = async { // action logic here};?/ or with parameters separationmodules { // action logic here};
Action usage
Calling action within other action
modules { // we can call other actions via context // assuming we define actions domain (see bellow for more examples) const results = await contextactions; return keyword: contextkeyword searchResults; results };
Calling the above action, assuming it is called search under domain 'actions'
modulesexports = async { const searchResults = await contextactions; // passing some parameters};
Context creation
There are two main components:
- properties are translated into
context.<property name>
access pattern - functions are translated into
context.<domain>.<action name>
const createContext = ; // inject/configure contextconst context =
Context extends Flow API, hence it allows to mix pub/sub with context based approach
modulesexports = async { const searchResults = await contextactions; // passing some parameters // publish it context;};// consume it in some other actionmodulesexports = async { const searchResults = await context; return searchResults ;};
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 = Flow;const flow = ; // create consumer componentconst consumer = flow;consumer; // define producer componentconst producer = flow;// publishproducer;
Shorter form for clarity:
// create consumer componentflow;
Consuming multiple events for the given topic
// create consumer componentflow// generate events;
Consuming events as a stream
const buffer = ;// create consumer streamconst stream = flow;stream;stream// generate some dataflow;flow;flow;flow;
Consuming events via reader
const Flow = Flow;const buffer = ;const flow = ;// create consumer streamconst reader = flow;// generate some dataflow;flow;flow;flow; { whiletrue const data = await readernext if data === undefined break; console; } ;
Consuming multiple topics in one short
// consume multiple topicsflow;flow;flow;
Using promise
// create consumer componentflow;
Multiple consumers, single producer
// create consumer componentflow;
Chaining actions, mixing, etc.
// NOTE: the order of consume/define does not matter// create consumer componentflow// start chain reaction here // lets produce multiple events via event emitter// validate// consume multiple topics// can consume inside consume// can generate multiple events using pub;// NOTE: we can consume first event via promise if we are not interested in the rest; // for debug you can listen to all eventsflow
Join flows together
const base = ;baseconst flow = base;flow;
You can also make them depend on each other
const base = ;base;const flow = base;flow;flow; // 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 // 300 ms;
Querying for the state
Oja provides a current status of topcis that have not been resolved
flow; console; // prints [foo, too]
Error
Throwing error
flow;// orflow;// orflow;// orflow;
Catching error
flow;// Orflow;
Error stops flow
The error will prevent further events including error events from publishing.
flow;
Composing complex flows
const base = ;base