工作流程引擎

概览

@datarster/workflow-engine 是用于Node.js的强大而可扩展的工作流程管理库。 它能够使用基于JSON的工作流程定义执行工作流程. 此引擎支持相继,平行,有条件,循环,开关等基于事件的工作流程,在出现故障时可选择回滚功能.

特征

• 支持各种工作流程步骤类型:任务、决定、平行、循环、等待、承诺、执行 当时 等待和切换.
• 事件驱动的执行,具有工作流程恢复能力.
• 与Redis合并,用于州管理和工作流程缓存.
• MongoDB 为工作流程执行历史进行记录 .
• 倒转支持失败步骤 与自定义倒转工人.
• 动态工人登记,提供超时支持.
• 有条件执行的切换/大小写逻辑.
• Redis 和 MongoDB 连接的灵活配置 .

安装

npm install @datarster/workflow-engine

使用量

导入和初始化引擎

import { WorkflowEngine } from "@datarster/workflow-engine";

const config = {
    redis: { host: "localhost", port: 6379 },
    useLogging: true,
    mongo: { uri: "mongodb://localhost:27017", database: "workflowDB", collection: "logs" }
};

const engine = new WorkflowEngine();
await engine.init(config);

定义工作流程

const workflow = {
    name: "Order Processing",
    steps: [
        { 
            id: "1", 
            type: "task", 
            name: "Validate Order", 
            worker: "validateOrder",
            workerTimeout: 30000,
            rollback: {
                id: "1_rollback",
                name: "Rollback Validation",
                type: "task",
                worker: "validateOrderRollback"
            }
        },
        { id: "2", type: "task", name: "Process Payment", worker: "processPayment" },
        { id: "3", type: "task", name: "Ship Order", worker: "shipOrder" }
    ]
};

await engine.registerWorkflow("order_processing", workflow);

登记工人

engine.registerWorker("validateOrder", async (data) => {
    console.log("Validating order", data);
    // Your validation logic here
});

engine.registerWorker("validateOrderRollback", async (data) => {
    console.log("Rolling back order validation", data);
    // Your rollback logic here
});

engine.registerWorker("processPayment", async (data) => {
    console.log("Processing payment", data);
    // Your payment logic here
});

engine.registerWorker("shipOrder", async (data) => {
    console.log("Shipping order", data);
    // Your shipping logic here
});

执行工作流程

const executionId = await engine.executeWorkflow({
    workflowId: "order_processing",
    inputData: { orderId: 12345 },
    callbackOptions: {
        onStepComplete: (step, result) => {
            console.log(`Step ${step.name} completed:`, result);
        },
        onError: (step, error) => {
            console.error(`Step ${step.name} failed:`, error);
        }
    }
});
console.log("Workflow execution started with ID:", executionId);

随事件恢复工作流程

// Resume a waiting workflow with an event
await engine.resumeWorkflow(executionId, "external_approval_received", {
    approved: true,
    approver: "[email protected]"
});

获取工作流程日志

const log = await engine.fetchWorkflowLog(executionId);
console.log("Execution Log:", log);

支持的工作流程步骤类型

公共 API 方法

以下方法可供参考: WorkflowEngine 类 :

说明: 这个 fetchWorkflowLog 方法在用法示例中显示,但在主类中没有被执行. 对于工作流程日志,请使用您自己的MongoDB查询或根据需要扩展引擎.

工作流程Step 接口( 高级)

每个工作流程步骤都支持以下领域:

• id (字符串):该步骤的独特标识符.
• type (字符串):步骤类型(参见所支持的工作流步骤类型).
• name (字符串):阶下名.
• worker (弦,可选): 要执行的工人函数名称 .
• condition (弦,可选):决定/跳跃步骤的条件.
• true_branch 编号 : false_branch (阵列,可选): 决策分支的步骤.
• parallel_steps (阵列,可选): 平行运行的步骤.
• switch_cases (阵列,可选): 切换步骤的案例.
• switch_case_steps (阵列,可选): 特定切换例的步骤 .
• event_name (字符串,可选):等待的事件(等待/执行 当时 等待).
• rollback (对象,可选): 步骤执行回滚 .
• status (弦,可选): 步骤状况.
• createdAt 编号 : updatedAt (字符串,可选):时间戳.
• workerTimeout (数目,可选): 工时超时数.
• field 编号 : value (任何,可选):用于开关/大小写逻辑.
• default_case (对象,可选): 切换的默认步骤 .

高级工作流程示例

切换/案件工作流程

const switchWorkflow = {
    name: "Order Status Processing",
    steps: [
        {
            id: "1",
            type: "switch",
            name: "Process Based on Status",
            switch_cases: [
                {
                    id: "case1",
                    field: "status",
                    value: "COMMITTED",
                    switch_case_steps: [
                        { id: "1.1", type: "task", name: "Handle Committed", worker: "handleCommitted" }
                    ]
                },
                {
                    id: "case2", 
                    field: "status",
                    value: "CANCELLED",
                    switch_case_steps: [
                        { id: "1.2", type: "task", name: "Handle Cancelled", worker: "handleCancelled" }
                    ]
                }
            ],
            default_case: {
                id: "default",
                type: "task",
                name: "Handle Default",
                worker: "handleDefault"
            }
        }
    ]
};

并行执行

const parallelWorkflow = {
    name: "Parallel Processing",
    steps: [
        {
            id: "1",
            type: "parallel",
            name: "Process Multiple Tasks",
            parallel_steps: [
                { id: "1.1", type: "task", name: "Task A", worker: "workerA" },
                { id: "1.2", type: "task", name: "Task B", worker: "workerB" },
                { id: "1.3", type: "task", name: "Task C", worker: "workerC" }
            ]
        }
    ]
};

事件驱动工作流程

const eventWorkflow = {
    name: "Event-Driven Process",
    steps: [
        { id: "1", type: "task", name: "Initial Task", worker: "initialWorker" },
        {
            id: "2",
            type: "wait",
            name: "Wait for Approval",
            event_name: "approval_received"
        },
        { id: "3", type: "task", name: "Final Task", worker: "finalWorker" }
    ]
};

配置选项

Redis 配置

const redisConfig = {
    host: "localhost",
    port: 6379,
    // OR provide your own Redis client
    redisClient: existingRedisClient
};

MongoDB 配置

const mongoConfig = {
    uri: "mongodb://localhost:27017",
    database: "workflowDB",
    collection: "logs",
    // OR provide your own MongoDB client
    mongoClient: existingMongoClient
};

处理和回滚出错

• 如果任务失败,工作流程引擎会触发回滚步骤,如果它们被定义的话.
• 回滚步骤按倒转顺序执行(LIFO - Last In, First Out).
• 倒滚步骤出错被记录,但不会停止倒滚进程.
• 每一步骤都可以在以下定义自己的回滚逻辑: rollback 属性.

州管理层

• 红: 用于缓存工作流程执行状态和已注册的工作流程.
• 蒙哥达: 存储详细的工作流程执行历史和日志(可选).
• 记忆中: 如果 Redis 没有配置, 则返回到内存中 .

类型脚本支持

库以 TypeScript 书写,并提供完整类型定义:

import { WorkflowEngine, WorkflowStepType, WorkflowDefinition } from "@datarster/workflow-engine";

捐款

欢迎捐款! 在GitHub上随意提交问题或调取请求.

许可证

麻省理工学院