DeFlow

Distributed, Decentralized Workflow creator for Node, Backed by Redis.

Philosophy

DeFlow attempt to fill the gap between a job scheduler and an ETL.
It manage workflow queue with 3 mains elements :

Workflow:

A Workflow defines a set of one or more steps, there created by a node process.

Steps:

Steps define a specific job that compose the workflow. They're treated sequentially and are described by a single file (module). This module file define the step lifecycle with predefined methods (before/after, error handler, task handler).

Each Step contain one or more tasks. A step can create one or more other steps, that way they can be multidimensional.

Tasks:

Tasks are treated by the step "handler" lifecycle method, they are designed to be treated concurrently between node. Tasks are handled functionnaly, accepting params and returning results. Tasks are configurable and can have timeout and retry strategy.

Main features

• Distributed: Intelligent distribution and parallelization of tasks between multiple nodeJS process.
• Decentralized: Designed to be crash proof, backed by Redis, pub-sub communication between nodes.
• Lifecycle: Lifecycle method allow you to manage and evolve the workflow process.
• Living workflow: Create steps or tasks on the fly, depending on the previous results.
• Promises based API.
• Configurable concurrency, retries, error handling and more.
• TypeScript support.

Getting started

install:

npm i deflow

declare a step handler:

// ./steps/string-to-number.ts import{StepHandler}from'deflow';/** * Declare the step handler and types * In this one, we convert string to float * NOTE: IT MUST BE EXPORTED AS DEFAULT */exportdefaultnewStepHandler({/** * Init method allow you to prepare tasks based on anything you want * @param step */asyncbeforeAll(step){consttasks=['12','10','7','45'];// You can fetch data from external source or dbawaitstep.addTasks(tasks);},/** * This method will run for each task of the step * @param task */asynchandler(task){returnparseFloat(task.data);},/** * This method is executed after each tasks done * Useful to log progress and stuff * @param task * @param step */asyncafterEach(task,step){constprogress=awaitstep.getProgress();console.log('Step1: afterEach',progress);console.log('Step1: Result',task.result);// Should be a floating number},/** * This method is executed after all tasks done * Useful to save results in a db or whatever you want * @param step */asyncafterAll(step){console.log('Step1: afterAll',awaitstep.getProgress());console.log('Step1: Result',awaitstep.getResults());},});

declare a workflow:

// ./index.ts importDeFlow,{WorkFlow}from'deflow';// Register deflow to your redis backendDeFlow.register({connection: {host: 'localhost',port: 6379}});/** * Workflow test file */functionrunWorkflow(){WorkFlow.create('some-custom-name').addStep({step: import('./steps/string-to-number')})// Register the step.addStep({step: import('./steps/anther-process-step')})// Register the step.run();// Run the workflow}// Run the workflow from somewhere in your code (make sure redis is ready before)setTimeout(()=>{runWorkflow();},2000)

The step attribute of addStep method can take a module, dynamic import, or a path to the module. Prefer the module or dynamic import to take advantage of TS type checking.

Check the complete API Documentation

Making things type safe

DeFlow allow you to define type safe stepHandler :

/** * Sending invoices for each user having purchased a product today */exportdefaultnewStepHandler<{date: Date},// step data type{productName: string;productPrice: number;userEmail: string;userName: string},// Task data type{sentStatus: boolean}// Task result type>({/** * Fetch user data from a database */asyncbeforeAll(step){constorders=awaitOrder.find({createdAt: step.data.date});// Create one task by orderconsttasks=orders.map((order)=>({productName: order.name,productPrice: order.price,userEmail: order.user.email,userName: order.user.name,}));returnstep.addTasks(tasks);// Each task will be processed by "handler" method},/** * For each task, send the invoice by email * Note that you can also access to the step "global" data * @param task * @param step */asynchandler(task,step){constsendMailRes=awaitsendmail({from: 'no-reply@yourdomain.com',to: task.data.userEmail,subject: `Your ${step.data.date} invoice for ${task.data.productName}`,html: ` Hello ${task.data.userName}, You just spent ${task.data.productPrice} for ${task.data.productName} `,});return{sentStatus: sendMailRes.status};},/** * After all task done, log some important things and take needed actions based on results * @param step */asyncafterAll(step){constres=awaitstep.getResults();consterrors=res.filter((task)=>task.result.sentStatus===false);if(errors.length>0){console.warn(`WARNING: ${errors.length} sendmail errors! Will try fallback method`);// Add another step right after this one when we have errorsawaitstep.addAfter({step: sendMailFallBackStep,tasks: errors,// If needed, you can directly add tasks while adding stepoptions: {taskConcurrency: 3,// Allow each node to treat 3 tasks concurrentlytaskTimeout: 2000,// Allow a timeout of 2000ms per tasktaskMaxFailCount: 3,// Allow a maximum of 3 retries},});}},});

Resources

• API Documentation
• Examples

Coming next

• More events that you can listen too
• Advanced concurrency management
• Reducer