mirror of https://github.com/Budibase/budibase.git
Andrew Kingston
6 years ago
5 changed files with 100 additions and 58 deletions
@ -1,67 +1,31 @@ |
|||
const mustache = require("mustache") |
|||
const actions = require("./actions") |
|||
const logic = require("./logic") |
|||
const triggers = require("./triggers") |
|||
const workerFarm = require("worker-farm") |
|||
const CouchDB = require("../db/client") |
|||
const singleThread = require("./thread") |
|||
|
|||
/** |
|||
* The workflow orchestrator is a class responsible for executing workflows. |
|||
* It relies on the strategy pattern, which allows composable behaviour to be |
|||
* passed into its execute() function. This allows custom execution behaviour based |
|||
* on where the orchestrator is run. |
|||
* |
|||
*/ |
|||
class Orchestrator { |
|||
constructor(workflow) { |
|||
this._context = {} |
|||
this._workflow = workflow |
|||
} |
|||
|
|||
async getStep(type, stepId) { |
|||
let step = null |
|||
if (type === "ACTION") { |
|||
step = await actions.getAction(stepId) |
|||
} else if (type === "LOGIC") { |
|||
step = logic.getLogic(stepId) |
|||
} |
|||
if (step == null) { |
|||
throw `Cannot find workflow step by name ${stepId}` |
|||
} |
|||
return step |
|||
} |
|||
|
|||
async execute(context) { |
|||
let workflow = this._workflow |
|||
if (!workflow.live) { |
|||
return |
|||
} |
|||
for (let block of workflow.definition.steps) { |
|||
let step = await this.getStep(block.type, block.stepId) |
|||
let args = { ...block.args } |
|||
// bind the workflow action args to the workflow context, if required
|
|||
for (let arg of Object.keys(args)) { |
|||
const argValue = args[arg] |
|||
// We don't want to render mustache templates on non-strings
|
|||
if (typeof argValue !== "string") continue |
|||
let workers = workerFarm(require.resolve("./thread")) |
|||
|
|||
args[arg] = mustache.render(argValue, { context: this._context }) |
|||
function runWorker(job) { |
|||
return new Promise((resolve, reject) => { |
|||
workers(job, err => { |
|||
if (err) { |
|||
reject(err) |
|||
} else { |
|||
resolve() |
|||
} |
|||
const response = await step({ |
|||
args, |
|||
context, |
|||
}) |
|||
|
|||
this._context = { |
|||
...this._context, |
|||
[block.id]: response, |
|||
} |
|||
} |
|||
} |
|||
}) |
|||
}) |
|||
} |
|||
|
|||
/** |
|||
* This module is built purely to kick off the worker farm and manage the inputs/outputs |
|||
*/ |
|||
module.exports.init = function() { |
|||
triggers.workflowQueue.process(async job => { |
|||
// Create orchestrator for each individual workflow (their own context)
|
|||
const workflowOrchestrator = new Orchestrator(job.data.workflow) |
|||
await workflowOrchestrator.execute(job.data.event) |
|||
if (CouchDB.preferredAdapters != null && CouchDB.preferredAdapters[0] !== "leveldb") { |
|||
await runWorker(job) |
|||
} else { |
|||
await singleThread(job) |
|||
} |
|||
}) |
|||
} |
|||
|
|||
@ -0,0 +1,67 @@ |
|||
const mustache = require("mustache") |
|||
const actions = require("./actions") |
|||
const logic = require("./logic") |
|||
|
|||
/** |
|||
* The workflow orchestrator is a class responsible for executing workflows. |
|||
* It handles the context of the workflow and makes sure each step gets the correct |
|||
* inputs and handles any outputs. |
|||
*/ |
|||
class Orchestrator { |
|||
constructor(workflow) { |
|||
this._context = {} |
|||
this._workflow = workflow |
|||
} |
|||
|
|||
async getStep(type, stepId) { |
|||
let step = null |
|||
if (type === "ACTION") { |
|||
step = await actions.getAction(stepId) |
|||
} else if (type === "LOGIC") { |
|||
step = logic.getLogic(stepId) |
|||
} |
|||
if (step == null) { |
|||
throw `Cannot find workflow step by name ${stepId}` |
|||
} |
|||
return step |
|||
} |
|||
|
|||
async execute(context) { |
|||
let workflow = this._workflow |
|||
for (let block of workflow.definition.steps) { |
|||
let step = await this.getStep(block.type, block.stepId) |
|||
let args = { ...block.args } |
|||
// bind the workflow action args to the workflow context, if required
|
|||
for (let arg of Object.keys(args)) { |
|||
const argValue = args[arg] |
|||
// We don't want to render mustache templates on non-strings
|
|||
if (typeof argValue !== "string") continue |
|||
|
|||
args[arg] = mustache.render(argValue, { context: this._context }) |
|||
} |
|||
const response = await step({ |
|||
args, |
|||
context, |
|||
}) |
|||
|
|||
this._context = { |
|||
...this._context, |
|||
[block.id]: response, |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
module.exports = async (job, cb = null) => { |
|||
try { |
|||
const workflowOrchestrator = new Orchestrator(job.data.workflow) |
|||
await workflowOrchestrator.execute(job.data.event) |
|||
if (cb) { |
|||
cb() |
|||
} |
|||
} catch (err) { |
|||
if (cb) { |
|||
cb(err) |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue