microflow

Welcome to microflow 👋

Finite state machine based HTTP microservice orchestration

🏠 Homepage

Purpose

Microflow helps you build and run complex workflows which are composed of HTTP microservices and manual (human moderator) stages, all by definiing a JSON workflow blueprint. It is built on robust concepts of finite state machine, and allows you to control input/output data as template variables (think jsonpath, handlebars). A workflow consists of manual states and task states (aka HTTP workers which could be sync/async).

Install

npm

npm i --save microflow

Documentation

The Microflow class provides various methods to author/execute/infer workflow and workflow instances

import { Microflow } from "microflow";
const flow = new Microflow();
const {
  // create task
  createTask,
  // create/put workflow
  createWorkflow,
  // get task
  getTask,
  // get workflow
  getWorkflow,

  // start workflow instance
  startWorkflow,
  // send event to workflow instance
  sendEvent,
  // get workflow instance details (useful for UI)
  getWorkflowInstance,

  // update task
  updateTask
  // update workflow
  updateWorkflow
  // delete task
  deleteTask
  // delete workflow
  deleteWorkflow
} = flow;

Usage

import { Microflow } from "microflow";

const flow = new Microflow({
  // bring your own storage connector for persistence
  storage: null
});

// Authoring task and workflow

// Register a task
const task = await flow.createTask({
  // type of task (only 'http' supported right now)
  type: "http",
  //  <AxiosRequestConfig> supported (https://github.com/axios/axios/blob/master/index.d.ts#L44)
  config: {
    url: "http://localhost:1000/api/experimental/dags/{{dagId}}/dag_runs",
    headers: {
      "Cache-Control": "no-cache",
      "Content-Type": "application/json"
    },
    data: "$.data",
    method: "post"
  }
});

// Register a workflow
const workflow = await flow.createWorkflow({
  definition: {
    initial: "waiting",
    states: {
      waiting: {
        on: {
          start_pipeline: {
            target: "start_dag",
          },
        },
      },
      start_dag: {
        invoke: {
          src: {
            type: "task",
            taskId: task.id,
            config: {
              parameters: {
                dagId: "dag1",
                data: "$",
              },
              resultSelector: {
                foo: "bar",
                baz: "har",
                message: "$.message",
                dag_execution_date: "$.execution_date",
              },
              resultPath: "$.pipeline.startDetails",
            },
          },
          onDone: {
            target: "pipeline_running",
          },
          onError: {
            target: "failed",
          },
        },
      },
      pipeline_running: {
        on: {
          test_complete: {
            target: "ready_for_approval",
            meta: {
              config: {
                resultPath: "$.pipeline.output",
              },
            },
          },
        },
      },
      ready_for_approval: {
        on: {
          approve: {
            target: "done",
          },
          reject: {
            target: "failed",
          },
        },
      },
      done: {
        type: "final",
      },
      failed: {
        type: "final",
      },
    },
  },
});

// Executing workflow aka "workflow instances"

const { id } = await flow.startWorkflow(workflow.id);

// Sending events to workflow instance
const response1 = await flow.sendEvent(id, {
    type: "start_pipeline"
});

// The above event will automatically fire the HTTP request configured in the task
/*
response1 = {
    "currentState": "pipeline_running",
    "completed": false,
    "nextEvents": [
        "test_complete"
    ]
}
*/

// Sending events to workflow instance
const response2 = await flow.sendEvent(id, {
    type: "test_complete",
    data: {
      kfold: 0.76,
      blind : 0.60
    }
});

/*
response2 = {
    "currentState": "ready_for_approval",
    "completed": false,
    "nextEvents": [
        "reject",
        "approve"
    ]
}
*/

const response3 = await flow.sendEvent(id, {
  type: "approve",
  data: {
      comment: "fair enough"
  }
});

/*
response3 = {
    "currentState": "done",
    "completed": true,
    "nextEvents": []
}
*/

Storage

import { Microflow } from "microflow";
import { MicroflowStorage } from "microflow/types";

// define your own storage from MicroflowStorage abstract class
class MyStorage implements MicroflowStorage {
  // define CRUD functions
}

const store = new MyStorage();

// create an instance of microflow with custom store injected
const flow = new Microflow({
  storage: store
});

Examples

Navigate to examples/basic to run a sample express project with ephemeral storage.

Run tests

npm run test

Author

👤 Karan Chhabra

• Github: @krn0x2
• LinkedIn: @krn0x2

🤝 Contributing

Contributions, issues and feature requests are welcome!
Feel free to check issues page. You can also take a look at the contributing guide.

Show your support

Give a ⭐️ if this project helped you!

📝 License

Copyright © 2020 Karan Chhabra.
This project is MIT licensed.