mohiit1502/Automata
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

✨ Separate webhooks from core (#1408)

Browse Source 
* Unify execution ID across executions

* Fix indentation and improved comments

* WIP: saving data after each node execution

* Added on/off to save data after each step, saving initial data and retries working

* Fixing lint issues

* Fixing more lint issues

* ✨ Add bull to execute workflows

* 👕 Fix lint issue

* ⚡ Add graceful shutdown to worker

* ⚡ Add loading staticData to worker

* 👕 Fix lint issue

* ⚡ Fix import

* Changed tables metadata to add nullable to stoppedAt

* Reload database on migration run

* Fixed reloading database schema for sqlite by reconnecting and fixing postgres migration

* Added checks to Redis and exiting process if connection is unavailable

* Fixing error with new installations

* Fix issue with data not being sent back to browser on manual executions with defined destination

* Merging bull and unify execution id branch fixes

* Main process will now get execution success from database instead of redis

* Omit execution duration if execution did not stop

* Fix issue with execution list displaying inconsistant information information while a workflow is running

* Remove unused hooks to clarify for developers that these wont run in queue mode

* Added active pooling to help recover from Redis crashes

* Lint issues

* Changing default polling interval to 60 seconds

* Removed unnecessary attributes from bull job

* Added webhooks service and setting to disable webhooks from main process

* Fixed executions list when running with queues. Now we get the list of actively running workflows from bull.

* Add option to disable deregistration of webhooks on shutdown

* Rename WEBHOOK_TUNNEL_URL to WEBHOOK_URL keeping backwards compat.

* Added auto refresh to executions list

* Improvements to workflow stop process when running with queues

* Refactor queue system to use a singleton and avoid code duplication

* Improve comments and remove unnecessary commits

* Remove console.log from vue file

* Blocking webhook process to run without queues

* Handling execution stop graciously when possible

* Removing initialization of all workflows from webhook process

* Refactoring code to remove code duplication for job stop

* Improved execution list to be more fluid and less intrusive

* Fixing workflow name for current executions when auto updating

* ⚡ Right align autorefresh checkbox

Co-authored-by: Jan Oberhauser <jan.oberhauser@gmail.com>
This commit is contained in:
Omar Ajoue
2021-02-09 23:32:40 +01:00
committed by GitHub
parent 98fa529e51
commit e53efdd337
18 changed files with 869 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/cli/src/ActiveWorkflowRunner.ts
View File

@@ -76,6 +76,10 @@ export class ActiveWorkflowRunner {
}
}
async initWebhooks() {
this.activeWorkflows = new ActiveWorkflows();
}
/**
* Removes all the currently active workflows
*

67
packages/cli/src/Queue.ts Normal file
View File

@@ -0,0 +1,67 @@
import * as Bull from 'bull';
import * as config from '../config';
import { IBullJobData } from './Interfaces';
export class Queue {
private jobQueue: Bull.Queue;
constructor() {
const prefix = config.get('queue.bull.prefix') as string;
const redisOptions = config.get('queue.bull.redis') as object;
// Disabling ready check is necessary as it allows worker to
// quickly reconnect to Redis if Redis crashes or is unreachable
// for some time. With it enabled, worker might take minutes to realize
// redis is back up and resume working.
// More here: https://github.com/OptimalBits/bull/issues/890
// @ts-ignore
this.jobQueue = new Bull('jobs', { prefix, redis: redisOptions, enableReadyCheck: false });
}
async add(jobData: IBullJobData, jobOptions: object): Promise<Bull.Job> {
return await this.jobQueue.add(jobData,jobOptions);
}
async getJob(jobId: Bull.JobId): Promise<Bull.Job | null> {
return await this.jobQueue.getJob(jobId);
}
async getJobs(jobTypes: Bull.JobStatus[]): Promise<Bull.Job[]> {
return await this.jobQueue.getJobs(jobTypes);
}
getBullObjectInstance(): Bull.Queue {
return this.jobQueue;
}
/**
*
* @param job A Bull.Job instance
* @returns boolean true if we were able to securely stop the job
*/
async stopJob(job: Bull.Job): Promise<boolean> {
if (await job.isActive()) {
// Job is already running so tell it to stop
await job.progress(-1);
return true;
} else {
// Job did not get started yet so remove from queue
try {
await job.remove();
return true;
} catch (e) {
await job.progress(-1);
}
}
return false;
}
}
let activeQueueInstance: Queue | undefined;
export function getInstance(): Queue {
if (activeQueueInstance === undefined) {
activeQueueInstance = new Queue();
}
return activeQueueInstance;
}

1
packages/cli/src/ResponseHelper.ts
View File

@@ -188,6 +188,7 @@ export function unflattenExecutionData(fullExecutionData: IExecutionFlattedDb):
startedAt: fullExecutionData.startedAt,
stoppedAt: fullExecutionData.stoppedAt,
finished: fullExecutionData.finished ? fullExecutionData.finished : false,
workflowId: fullExecutionData.workflowId,
});
return returnData;

232
packages/cli/src/Server.ts
View File

@@ -62,6 +62,7 @@ import {
ResponseHelper,
TestWebhooks,
WebhookHelpers,
WebhookServer,
WorkflowCredentials,
WorkflowExecuteAdditionalData,
WorkflowRunner,
@@ -105,6 +106,7 @@ import * as jwks from 'jwks-rsa';
import * as timezones from 'google-timezones-json';
import * as parseUrl from 'parseurl';
import * as querystring from 'querystring';
import * as Queue from '../src/Queue';
import { OptionsWithUrl } from 'request-promise-native';
class App {
@@ -1428,7 +1430,14 @@ class App {
limit = parseInt(req.query.limit as string, 10);
}
const executingWorkflowIds = this.activeExecutionsInstance.getActiveExecutions().map(execution => execution.id.toString()) as string[];
let executingWorkflowIds;
if (config.get('executions.mode') === 'queue') {
const currentJobs = await Queue.getInstance().getJobs(['active', 'waiting']);
executingWorkflowIds = currentJobs.map(job => job.data.executionId) as string[];
} else {
executingWorkflowIds = this.activeExecutionsInstance.getActiveExecutions().map(execution => execution.id.toString()) as string[];
}
const countFilter = JSON.parse(JSON.stringify(filter));
countFilter.select = ['id'];
@@ -1453,10 +1462,10 @@ class App {
resultsQuery.andWhere(`execution.${filterField} = :${filterField}`, {[filterField]: filter[filterField]});
});
if (req.query.lastId) {
resultsQuery.andWhere(`execution.id <= :lastId`, {lastId: req.query.lastId});
resultsQuery.andWhere(`execution.id < :lastId`, {lastId: req.query.lastId});
}
if (req.query.firstId) {
resultsQuery.andWhere(`execution.id >= :firstId`, {firstId: req.query.firstId});
resultsQuery.andWhere(`execution.id > :firstId`, {firstId: req.query.firstId});
}
if (executingWorkflowIds.length > 0) {
resultsQuery.andWhere(`execution.id NOT IN (:...ids)`, {ids: executingWorkflowIds});
@@ -1626,52 +1635,114 @@ class App {
// Returns all the currently working executions
this.app.get(`/${this.restEndpoint}/executions-current`, ResponseHelper.send(async (req: express.Request, res: express.Response): Promise<IExecutionsSummary[]> => {
const executingWorkflows = this.activeExecutionsInstance.getActiveExecutions();
if (config.get('executions.mode') === 'queue') {
const currentJobs = await Queue.getInstance().getJobs(['active', 'waiting']);
const returnData: IExecutionsSummary[] = [];
const currentlyRunningExecutionIds = currentJobs.map(job => job.data.executionId);
let filter: any = {}; // tslint:disable-line:no-any
if (req.query.filter) {
filter = JSON.parse(req.query.filter as string);
}
for (const data of executingWorkflows) {
if (filter.workflowId !== undefined && filter.workflowId !== data.workflowId) {
continue;
}
returnData.push(
{
idActive: data.id.toString(),
workflowId: data.workflowId.toString(),
mode: data.mode,
retryOf: data.retryOf,
startedAt: new Date(data.startedAt),
const resultsQuery = await Db.collections.Execution!
.createQueryBuilder("execution")
.select([
'execution.id',
'execution.workflowId',
'execution.mode',
'execution.retryOf',
'execution.startedAt',
])
.orderBy('execution.id', 'DESC')
.andWhere(`execution.id IN (:...ids)`, {ids: currentlyRunningExecutionIds});
if (req.query.filter) {
const filter = JSON.parse(req.query.filter as string);
if (filter.workflowId !== undefined) {
resultsQuery.andWhere('execution.workflowId = :workflowId', {workflowId: filter.workflowId});
}
);
}
}
return returnData;
const results = await resultsQuery.getMany();
return results.map(result => {
return {
idActive: result.id,
workflowId: result.workflowId,
mode: result.mode,
retryOf: result.retryOf !== null ? result.retryOf : undefined,
startedAt: new Date(result.startedAt),
} as IExecutionsSummary;
});
} else {
const executingWorkflows = this.activeExecutionsInstance.getActiveExecutions();
const returnData: IExecutionsSummary[] = [];
let filter: any = {}; // tslint:disable-line:no-any
if (req.query.filter) {
filter = JSON.parse(req.query.filter as string);
}
for (const data of executingWorkflows) {
if (filter.workflowId !== undefined && filter.workflowId !== data.workflowId) {
continue;
}
returnData.push(
{
idActive: data.id.toString(),
workflowId: data.workflowId.toString(),
mode: data.mode,
retryOf: data.retryOf,
startedAt: new Date(data.startedAt),
}
);
}
return returnData;
}
}));
// Forces the execution to stop
this.app.post(`/${this.restEndpoint}/executions-current/:id/stop`, ResponseHelper.send(async (req: express.Request, res: express.Response): Promise<IExecutionsStopData> => {
const executionId = req.params.id;
if (config.get('executions.mode') === 'queue') {
const currentJobs = await Queue.getInstance().getJobs(['active', 'waiting']);
// Stopt he execution and wait till it is done and we got the data
const result = await this.activeExecutionsInstance.stopExecution(executionId);
const job = currentJobs.find(job => job.data.executionId.toString() === req.params.id);
if (result === undefined) {
throw new Error(`The execution id "${executionId}" could not be found.`);
if (!job) {
throw new Error(`Could not stop "${req.params.id}" as it is no longer in queue.`);
} else {
await Queue.getInstance().stopJob(job);
}
const executionDb = await Db.collections.Execution?.findOne(req.params.id) as IExecutionFlattedDb;
const fullExecutionData = ResponseHelper.unflattenExecutionData(executionDb) as IExecutionResponse;
const returnData: IExecutionsStopData = {
mode: fullExecutionData.mode,
startedAt: new Date(fullExecutionData.startedAt),
stoppedAt: fullExecutionData.stoppedAt ? new Date(fullExecutionData.stoppedAt) : undefined,
finished: fullExecutionData.finished,
};
return returnData;
} else {
const executionId = req.params.id;
// Stopt he execution and wait till it is done and we got the data
const result = await this.activeExecutionsInstance.stopExecution(executionId);
if (result === undefined) {
throw new Error(`The execution id "${executionId}" could not be found.`);
}
const returnData: IExecutionsStopData = {
mode: result.mode,
startedAt: new Date(result.startedAt),
stoppedAt: result.stoppedAt ? new Date(result.stoppedAt) : undefined,
finished: result.finished,
};
return returnData;
}
const returnData: IExecutionsStopData = {
mode: result.mode,
startedAt: new Date(result.startedAt),
stoppedAt: result.stoppedAt ? new Date(result.stoppedAt) : undefined,
finished: result.finished,
};
return returnData;
}));
@@ -1711,88 +1782,9 @@ class App {
// Webhooks
// ----------------------------------------
// HEAD webhook requests
this.app.head(`/${this.endpointWebhook}/*`, async (req: express.Request, res: express.Response) => {
// Cut away the "/webhook/" to get the registred part of the url
const requestUrl = (req as ICustomRequest).parsedUrl!.pathname!.slice(this.endpointWebhook.length + 2);
let response;
try {
response = await this.activeWorkflowRunner.executeWebhook('HEAD', requestUrl, req, res);
} catch (error) {
ResponseHelper.sendErrorResponse(res, error);
return;
}
if (response.noWebhookResponse === true) {
// Nothing else to do as the response got already sent
return;
}
ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode);
});
// OPTIONS webhook requests
this.app.options(`/${this.endpointWebhook}/*`, async (req: express.Request, res: express.Response) => {
// Cut away the "/webhook/" to get the registred part of the url
const requestUrl = (req as ICustomRequest).parsedUrl!.pathname!.slice(this.endpointWebhook.length + 2);
let allowedMethods: string[];
try {
allowedMethods = await this.activeWorkflowRunner.getWebhookMethods(requestUrl);
allowedMethods.push('OPTIONS');
// Add custom "Allow" header to satisfy OPTIONS response.
res.append('Allow', allowedMethods);
} catch (error) {
ResponseHelper.sendErrorResponse(res, error);
return;
}
ResponseHelper.sendSuccessResponse(res, {}, true, 204);
});
// GET webhook requests
this.app.get(`/${this.endpointWebhook}/*`, async (req: express.Request, res: express.Response) => {
// Cut away the "/webhook/" to get the registred part of the url
const requestUrl = (req as ICustomRequest).parsedUrl!.pathname!.slice(this.endpointWebhook.length + 2);
let response;
try {
response = await this.activeWorkflowRunner.executeWebhook('GET', requestUrl, req, res);
} catch (error) {
ResponseHelper.sendErrorResponse(res, error);
return;
}
if (response.noWebhookResponse === true) {
// Nothing else to do as the response got already sent
return;
}
ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode);
});
// POST webhook requests
this.app.post(`/${this.endpointWebhook}/*`, async (req: express.Request, res: express.Response) => {
// Cut away the "/webhook/" to get the registred part of the url
const requestUrl = (req as ICustomRequest).parsedUrl!.pathname!.slice(this.endpointWebhook.length + 2);
let response;
try {
response = await this.activeWorkflowRunner.executeWebhook('POST', requestUrl, req, res);
} catch (error) {
ResponseHelper.sendErrorResponse(res, error);
return;
}
if (response.noWebhookResponse === true) {
// Nothing else to do as the response got already sent
return;
}
ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode);
});
if (config.get('endpoints.disableProductionWebhooksOnMainProcess') !== true) {
WebhookServer.registerProductionWebhooks.apply(this);
}
// HEAD webhook requests (test for UI)
this.app.head(`/${this.endpointWebhookTest}/*`, async (req: express.Request, res: express.Response) => {

7
packages/cli/src/WebhookHelpers.ts
View File

@@ -453,8 +453,11 @@ export function getWorkflowWebhooksBasic(workflow: Workflow): IWebhookData[] {
export function getWebhookBaseUrl() {
let urlBaseWebhook = GenericHelpers.getBaseUrl();
if (process.env.WEBHOOK_TUNNEL_URL !== undefined) {
urlBaseWebhook = process.env.WEBHOOK_TUNNEL_URL;
// We renamed WEBHOOK_TUNNEL_URL to WEBHOOK_URL. This is here to maintain
// backward compatibility. Will be deprecated and removed in the future.
if (process.env.WEBHOOK_TUNNEL_URL !== undefined || process.env.WEBHOOK_URL !== undefined) {
// @ts-ignore
urlBaseWebhook = process.env.WEBHOOK_TUNNEL_URL || process.env.WEBHOOK_URL;
}
return urlBaseWebhook;

306
packages/cli/src/WebhookServer.ts Normal file
View File

@@ -0,0 +1,306 @@
import * as express from 'express';
import {
readFileSync,
} from 'fs';
import {
getConnectionManager,
} from 'typeorm';
import * as bodyParser from 'body-parser';
require('body-parser-xml')(bodyParser);
import * as _ from 'lodash';
import {
ActiveExecutions,
ActiveWorkflowRunner,
Db,
ExternalHooks,
GenericHelpers,
ICustomRequest,
IExternalHooksClass,
IPackageVersions,
ResponseHelper,
} from './';
import * as compression from 'compression';
import * as config from '../config';
import * as parseUrl from 'parseurl';
export function registerProductionWebhooks() {
// HEAD webhook requests
this.app.head(`/${this.endpointWebhook}/*`, async (req: express.Request, res: express.Response) => {
// Cut away the "/webhook/" to get the registred part of the url
const requestUrl = (req as ICustomRequest).parsedUrl!.pathname!.slice(this.endpointWebhook.length + 2);
let response;
try {
response = await this.activeWorkflowRunner.executeWebhook('HEAD', requestUrl, req, res);
} catch (error) {
ResponseHelper.sendErrorResponse(res, error);
return;
}
if (response.noWebhookResponse === true) {
// Nothing else to do as the response got already sent
return;
}
ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode);
});
// OPTIONS webhook requests
this.app.options(`/${this.endpointWebhook}/*`, async (req: express.Request, res: express.Response) => {
// Cut away the "/webhook/" to get the registred part of the url
const requestUrl = (req as ICustomRequest).parsedUrl!.pathname!.slice(this.endpointWebhook.length + 2);
let allowedMethods: string[];
try {
allowedMethods = await this.activeWorkflowRunner.getWebhookMethods(requestUrl);
allowedMethods.push('OPTIONS');
// Add custom "Allow" header to satisfy OPTIONS response.
res.append('Allow', allowedMethods);
} catch (error) {
ResponseHelper.sendErrorResponse(res, error);
return;
}
ResponseHelper.sendSuccessResponse(res, {}, true, 204);
});
// GET webhook requests
this.app.get(`/${this.endpointWebhook}/*`, async (req: express.Request, res: express.Response) => {
// Cut away the "/webhook/" to get the registred part of the url
const requestUrl = (req as ICustomRequest).parsedUrl!.pathname!.slice(this.endpointWebhook.length + 2);
let response;
try {
response = await this.activeWorkflowRunner.executeWebhook('GET', requestUrl, req, res);
} catch (error) {
ResponseHelper.sendErrorResponse(res, error);
return;
}
if (response.noWebhookResponse === true) {
// Nothing else to do as the response got already sent
return;
}
ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode);
});
// POST webhook requests
this.app.post(`/${this.endpointWebhook}/*`, async (req: express.Request, res: express.Response) => {
// Cut away the "/webhook/" to get the registred part of the url
const requestUrl = (req as ICustomRequest).parsedUrl!.pathname!.slice(this.endpointWebhook.length + 2);
let response;
try {
response = await this.activeWorkflowRunner.executeWebhook('POST', requestUrl, req, res);
} catch (error) {
ResponseHelper.sendErrorResponse(res, error);
return;
}
if (response.noWebhookResponse === true) {
// Nothing else to do as the response got already sent
return;
}
ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode);
});
}
class App {
app: express.Application;
activeWorkflowRunner: ActiveWorkflowRunner.ActiveWorkflowRunner;
endpointWebhook: string;
endpointPresetCredentials: string;
externalHooks: IExternalHooksClass;
saveDataErrorExecution: string;
saveDataSuccessExecution: string;
saveManualExecutions: boolean;
executionTimeout: number;
maxExecutionTimeout: number;
timezone: string;
activeExecutionsInstance: ActiveExecutions.ActiveExecutions;
versions: IPackageVersions | undefined;
restEndpoint: string;
protocol: string;
sslKey: string;
sslCert: string;
presetCredentialsLoaded: boolean;
constructor() {
this.app = express();
this.endpointWebhook = config.get('endpoints.webhook') as string;
this.saveDataErrorExecution = config.get('executions.saveDataOnError') as string;
this.saveDataSuccessExecution = config.get('executions.saveDataOnSuccess') as string;
this.saveManualExecutions = config.get('executions.saveDataManualExecutions') as boolean;
this.executionTimeout = config.get('executions.timeout') as number;
this.maxExecutionTimeout = config.get('executions.maxTimeout') as number;
this.timezone = config.get('generic.timezone') as string;
this.restEndpoint = config.get('endpoints.rest') as string;
this.activeWorkflowRunner = ActiveWorkflowRunner.getInstance();
this.activeExecutionsInstance = ActiveExecutions.getInstance();
this.protocol = config.get('protocol');
this.sslKey = config.get('ssl_key');
this.sslCert = config.get('ssl_cert');
this.externalHooks = ExternalHooks();
this.presetCredentialsLoaded = false;
this.endpointPresetCredentials = config.get('credentials.overwrite.endpoint') as string;
}
/**
* Returns the current epoch time
*
* @returns {number}
* @memberof App
*/
getCurrentDate(): Date {
return new Date();
}
async config(): Promise<void> {
this.versions = await GenericHelpers.getVersions();
// Compress the response data
this.app.use(compression());
// Make sure that each request has the "parsedUrl" parameter
this.app.use((req: express.Request, res: express.Response, next: express.NextFunction) => {
(req as ICustomRequest).parsedUrl = parseUrl(req);
// @ts-ignore
req.rawBody = Buffer.from('', 'base64');
next();
});
// Support application/json type post data
this.app.use(bodyParser.json({
limit: '16mb', verify: (req, res, buf) => {
// @ts-ignore
req.rawBody = buf;
},
}));
// Support application/xml type post data
// @ts-ignore
this.app.use(bodyParser.xml({
limit: '16mb', xmlParseOptions: {
normalize: true, // Trim whitespace inside text nodes
normalizeTags: true, // Transform tags to lowercase
explicitArray: false, // Only put properties in array if length > 1
},
}));
this.app.use(bodyParser.text({
limit: '16mb', verify: (req, res, buf) => {
// @ts-ignore
req.rawBody = buf;
},
}));
//support application/x-www-form-urlencoded post data
this.app.use(bodyParser.urlencoded({ extended: false,
verify: (req, res, buf) => {
// @ts-ignore
req.rawBody = buf;
},
}));
if (process.env['NODE_ENV'] !== 'production') {
this.app.use((req: express.Request, res: express.Response, next: express.NextFunction) => {
// Allow access also from frontend when developing
res.header('Access-Control-Allow-Origin', 'http://localhost:8080');
res.header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS, PUT, PATCH, DELETE');
res.header('Access-Control-Allow-Headers', 'Origin, X-Requested-With, Content-Type, Accept, sessionid');
next();
});
}
this.app.use((req: express.Request, res: express.Response, next: express.NextFunction) => {
if (Db.collections.Workflow === null) {
const error = new ResponseHelper.ResponseError('Database is not ready!', undefined, 503);
return ResponseHelper.sendErrorResponse(res, error);
}
next();
});
// ----------------------------------------
// Healthcheck
// ----------------------------------------
// Does very basic health check
this.app.get('/healthz', async (req: express.Request, res: express.Response) => {
const connectionManager = getConnectionManager();
if (connectionManager.connections.length === 0) {
const error = new ResponseHelper.ResponseError('No Database connection found!', undefined, 503);
return ResponseHelper.sendErrorResponse(res, error);
}
if (connectionManager.connections[0].isConnected === false) {
// Connection is not active
const error = new ResponseHelper.ResponseError('Database connection not active!', undefined, 503);
return ResponseHelper.sendErrorResponse(res, error);
}
// Everything fine
const responseData = {
status: 'ok',
};
ResponseHelper.sendSuccessResponse(res, responseData, true, 200);
});
registerProductionWebhooks.apply(this);
}
}
export async function start(): Promise<void> {
const PORT = config.get('port');
const ADDRESS = config.get('listen_address');
const app = new App();
await app.config();
let server;
if (app.protocol === 'https' && app.sslKey && app.sslCert) {
const https = require('https');
const privateKey = readFileSync(app.sslKey, 'utf8');
const cert = readFileSync(app.sslCert, 'utf8');
const credentials = { key: privateKey, cert };
server = https.createServer(credentials, app.app);
} else {
const http = require('http');
server = http.createServer(app.app);
}
server.listen(PORT, ADDRESS, async () => {
const versions = await GenericHelpers.getVersions();
console.log(`n8n ready on ${ADDRESS}, port ${PORT}`);
console.log(`Version: ${versions.cli}`);
await app.externalHooks.run('n8n.ready', [app]);
});
}

24
packages/cli/src/WorkflowExecuteAdditionalData.ts
View File

@@ -193,20 +193,24 @@ function hookFunctionsPush(): IWorkflowExecuteHooks {
workflowExecuteBefore: [
async function (this: WorkflowHooks): Promise<void> {
// Push data to editor-ui once workflow finished
const pushInstance = Push.getInstance();
pushInstance.send('executionStarted', {
executionId: this.executionId,
mode: this.mode,
startedAt: new Date(),
retryOf: this.retryOf,
workflowId: this.workflowData.id as string,
workflowName: this.workflowData.name,
});
if (this.mode === 'manual') {
const pushInstance = Push.getInstance();
pushInstance.send('executionStarted', {
executionId: this.executionId,
mode: this.mode,
startedAt: new Date(),
retryOf: this.retryOf,
workflowId: this.workflowData.id as string,
workflowName: this.workflowData.name,
});
}
},
],
workflowExecuteAfter: [
async function (this: WorkflowHooks, fullRunData: IRun, newStaticData: IDataObject): Promise<void> {
pushExecutionFinished(this.mode, fullRunData, this.executionId, undefined, this.retryOf);
if (this.mode === 'manual') {
pushExecutionFinished(this.mode, fullRunData, this.executionId, undefined, this.retryOf);
}
},
],
};

44
packages/cli/src/WorkflowRunner.ts
View File

@@ -41,6 +41,7 @@ import { join as pathJoin } from 'path';
import { fork } from 'child_process';
import * as Bull from 'bull';
import * as Queue from './Queue';
export class WorkflowRunner {
activeExecutions: ActiveExecutions.ActiveExecutions;
@@ -57,11 +58,7 @@ export class WorkflowRunner {
const executionsMode = config.get('executions.mode') as string;
if (executionsMode === 'queue') {
// Connect to bull-queue
const prefix = config.get('queue.bull.prefix') as string;
const redisOptions = config.get('queue.bull.redis') as object;
// @ts-ignore
this.jobQueue = new Bull('jobs', { prefix, redis: redisOptions, enableReadyCheck: false });
this.jobQueue = Queue.getInstance().getBullObjectInstance();
}
}
@@ -251,30 +248,23 @@ export class WorkflowRunner {
const workflowExecution: PCancelable<IRun> = new PCancelable(async (resolve, reject, onCancel) => {
onCancel.shouldReject = false;
onCancel(async () => {
if (await job.isActive()) {
// Job is already running so tell it to stop
await job.progress(-1);
} else {
// Job did not get started yet so remove from queue
await job.remove();
await Queue.getInstance().stopJob(job);
const fullRunData: IRun = {
data: {
resultData: {
error: {
message: 'Workflow has been canceled!',
} as IExecutionError,
runData: {},
},
const fullRunData :IRun = {
data: {
resultData: {
error: {
message: 'Workflow has been canceled!',
} as IExecutionError,
runData: {},
},
mode: data.executionMode,
startedAt: new Date(),
stoppedAt: new Date(),
};
this.activeExecutions.remove(executionId, fullRunData);
resolve(fullRunData);
}
},
mode: data.executionMode,
startedAt: new Date(),
stoppedAt: new Date(),
};
this.activeExecutions.remove(executionId, fullRunData);
resolve(fullRunData);
});
const jobData: Promise<IBullJobResponse> = job.finished();

2
packages/cli/src/index.ts
View File

@@ -17,6 +17,7 @@ import * as ResponseHelper from './ResponseHelper';
import * as Server from './Server';
import * as TestWebhooks from './TestWebhooks';
import * as WebhookHelpers from './WebhookHelpers';
import * as WebhookServer from './WebhookServer';
import * as WorkflowExecuteAdditionalData from './WorkflowExecuteAdditionalData';
import * as WorkflowHelpers from './WorkflowHelpers';
export {
@@ -29,6 +30,7 @@ export {
Server,
TestWebhooks,
WebhookHelpers,
WebhookServer,
WorkflowExecuteAdditionalData,
WorkflowHelpers,
};