refactor(plugin-workflow): refactor request instruction (#1356)

This commit is contained in:
Junyi 2023-01-12 18:22:06 +08:00 committed by GitHub
parent 3e90bc9d7e
commit a1127300ae
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 104 additions and 140 deletions

View File

@ -12,6 +12,7 @@
"@nocobase/server": "0.9.0-alpha.1", "@nocobase/server": "0.9.0-alpha.1",
"@nocobase/utils": "0.9.0-alpha.1", "@nocobase/utils": "0.9.0-alpha.1",
"antd": "4.22.8", "antd": "4.22.8",
"axios": "^0.27.2",
"classnames": "^2.3.1", "classnames": "^2.3.1",
"cron-parser": "4.4.0", "cron-parser": "4.4.0",
"ejs": "^3.1.8", "ejs": "^3.1.8",

View File

@ -176,12 +176,12 @@ export default class Processor {
public async run(node, input?) { public async run(node, input?) {
const { instructions } = this.options.plugin; const { instructions } = this.options.plugin;
const { run } = instructions.get(node.type); const instruction = instructions.get(node.type);
if (typeof run !== 'function') { if (typeof instruction.run !== 'function') {
return Promise.reject(new Error('`run` should be implemented for customized execution of the node')); return Promise.reject(new Error('`run` should be implemented for customized execution of the node'));
} }
return this.exec(run, node, input); return this.exec(instruction.run.bind(instruction), node, input);
} }
// parent node should take over the control // parent node should take over the control
@ -200,12 +200,12 @@ export default class Processor {
async recall(node, job) { async recall(node, job) {
const { instructions } = this.options.plugin; const { instructions } = this.options.plugin;
const { resume } = instructions.get(node.type); const instruction = instructions.get(node.type);
if (typeof resume !== 'function') { if (typeof instruction.resume !== 'function') {
return Promise.reject(new Error('`resume` should be implemented')); return Promise.reject(new Error('`resume` should be implemented'));
} }
return this.exec(resume, node, job); return this.exec(instruction.resume.bind(instruction), node, job);
} }
async exit(job: JobModel | null) { async exit(job: JobModel | null) {

View File

@ -1,37 +1,14 @@
/**
* @jest-environment node
*/
import { Application } from '@nocobase/server'; import { Application } from '@nocobase/server';
import Database from '@nocobase/database'; import Database from '@nocobase/database';
import { getApp, sleep } from '..'; import { getApp, sleep } from '..';
import { RequestConfig } from '../../instructions/request'; import { RequestConfig } from '../../instructions/request';
import axios, { AxiosRequestConfig } from 'axios';
import { JOB_STATUS } from '../../constants'; import { JOB_STATUS } from '../../constants';
const testUrl = 'https://nocobase.com/test'; const PORT = 12345;
const url_400 = 'https://nocobase.com/400';
const timeoutUrl = 'https://nocobase.com/timeout'; const URL_DATA = `http://localhost:${PORT}/data`;
jest.mock('axios', () => { const URL_400 = `http://localhost:${PORT}/api/400`;
return { const URL_TIMEOUT = `http://localhost:${PORT}/timeout`;
request: async (config: AxiosRequestConfig) => {
await sleep(1000);
if (config.url === url_400) {
return {
data: config,
status: 400,
};
}
if (config.url === timeoutUrl) {
throw new Error('timeout');
}
return {
data: config,
status: 200,
};
},
};
});
describe('workflow > instructions > request', () => { describe('workflow > instructions > request', () => {
let app: Application; let app: Application;
@ -41,7 +18,27 @@ describe('workflow > instructions > request', () => {
let workflow; let workflow;
beforeEach(async () => { beforeEach(async () => {
app = await getApp(); app = await getApp({ manual: true });
app.use(async (ctx, next) => {
if (ctx.path === '/api/400') {
return ctx.throw(400);
}
if (ctx.path === '/timeout') {
await sleep(2000);
return ctx.throw(new Error('timeout'));
}
if (ctx.path === '/data') {
ctx.withoutDataWrapping = true;
ctx.body = {
meta: { title: ctx.query.title },
data: { title: ctx.request.body.title }
};
}
next();
});
await app.start({ listen: { port: PORT } });
db = app.db; db = app.db;
WorkflowModel = db.getCollection('workflows').model; WorkflowModel = db.getCollection('workflows').model;
@ -65,7 +62,7 @@ describe('workflow > instructions > request', () => {
await workflow.createNode({ await workflow.createNode({
type: 'request', type: 'request',
config: { config: {
url: 'https://www.baidu.com?name=lily&age=20', url: URL_DATA,
method: 'GET', method: 'GET',
} as RequestConfig, } as RequestConfig,
}); });
@ -76,12 +73,6 @@ describe('workflow > instructions > request', () => {
let [execution] = await workflow.getExecutions(); let [execution] = await workflow.getExecutions();
let [job] = await execution.getJobs(); let [job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.PENDING);
await sleep(2000);
[execution] = await workflow.getExecutions();
[job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.RESOLVED); expect(job.status).toEqual(JOB_STATUS.RESOLVED);
}); });
@ -89,7 +80,7 @@ describe('workflow > instructions > request', () => {
await workflow.createNode({ await workflow.createNode({
type: 'request', type: 'request',
config: { config: {
url: 'https://www.xxx.com?title=<%= ctx.data.title // 测试 %>', url: `${URL_DATA}?title=<%= ctx.data.title %>`,
method: 'GET', method: 'GET',
} as RequestConfig, } as RequestConfig,
}); });
@ -100,77 +91,69 @@ describe('workflow > instructions > request', () => {
let [execution] = await workflow.getExecutions(); let [execution] = await workflow.getExecutions();
let [job] = await execution.getJobs(); let [job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.PENDING);
await sleep(1000);
[execution] = await workflow.getExecutions();
[job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.RESOLVED); expect(job.status).toEqual(JOB_STATUS.RESOLVED);
expect(job.result.url).toEqual('https://www.xxx.com?title=t1'); console.log(job.result);
expect(job.result.meta.title).toEqual('t1');
}); });
it('request - timeout', async () => { it('request - timeout', async () => {
await workflow.createNode({ await workflow.createNode({
type: 'request', type: 'request',
config: { config: {
url: timeoutUrl, url: URL_TIMEOUT,
method: 'GET', method: 'GET',
timeout: 1000, timeout: 250,
} as RequestConfig, } as RequestConfig,
}); });
await PostRepo.create({ values: { title: 't1' } }); await PostRepo.create({ values: { title: 't1' } });
await sleep(500); await sleep(1000);
let [execution] = await workflow.getExecutions(); let [execution] = await workflow.getExecutions();
let [job] = await execution.getJobs(); let [job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.PENDING);
await sleep(1000);
[execution] = await workflow.getExecutions();
[job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.REJECTED); expect(job.status).toEqual(JOB_STATUS.REJECTED);
expect(job.result).toMatch('timeout');
expect(job.result).toMatchObject({
code: 'ECONNABORTED',
name: 'AxiosError',
status: null,
message: 'timeout of 250ms exceeded'
});
}); });
it('request - ignoreFail', async () => { it('request - ignoreFail', async () => {
await workflow.createNode({ await workflow.createNode({
type: 'request', type: 'request',
config: { config: {
url: timeoutUrl, url: URL_TIMEOUT,
method: 'GET', method: 'GET',
timeout: 1000, timeout: 250,
ignoreFail: true, ignoreFail: true,
} as RequestConfig, } as RequestConfig,
}); });
await PostRepo.create({ values: { title: 't1' } }); await PostRepo.create({ values: { title: 't1' } });
await sleep(500); await sleep(1000);
let [execution] = await workflow.getExecutions(); let [execution] = await workflow.getExecutions();
let [job] = await execution.getJobs(); let [job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.PENDING);
await sleep(1000);
[execution] = await workflow.getExecutions();
[job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.RESOLVED); expect(job.status).toEqual(JOB_STATUS.RESOLVED);
expect(job.result).toMatch('timeout'); expect(job.result).toMatchObject({
code: 'ECONNABORTED',
name: 'AxiosError',
status: null,
message: 'timeout of 250ms exceeded'
});
}); });
it('response 400', async () => { it('response 400', async () => {
await workflow.createNode({ await workflow.createNode({
type: 'request', type: 'request',
config: { config: {
url: url_400, url: URL_400,
method: 'GET', method: 'GET',
timeout: 1000,
ignoreFail: false, ignoreFail: false,
} as RequestConfig, } as RequestConfig,
}); });
@ -181,12 +164,6 @@ describe('workflow > instructions > request', () => {
let [execution] = await workflow.getExecutions(); let [execution] = await workflow.getExecutions();
let [job] = await execution.getJobs(); let [job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.PENDING);
await sleep(1000);
[execution] = await workflow.getExecutions();
[job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.REJECTED); expect(job.status).toEqual(JOB_STATUS.REJECTED);
expect(job.result.status).toBe(400); expect(job.result.status).toBe(400);
}); });
@ -195,7 +172,7 @@ describe('workflow > instructions > request', () => {
await workflow.createNode({ await workflow.createNode({
type: 'request', type: 'request',
config: { config: {
url: url_400, url: URL_400,
method: 'GET', method: 'GET',
timeout: 1000, timeout: 1000,
ignoreFail: true, ignoreFail: true,
@ -208,12 +185,6 @@ describe('workflow > instructions > request', () => {
let [execution] = await workflow.getExecutions(); let [execution] = await workflow.getExecutions();
let [job] = await execution.getJobs(); let [job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.PENDING);
await sleep(1000);
[execution] = await workflow.getExecutions();
[job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.RESOLVED); expect(job.status).toEqual(JOB_STATUS.RESOLVED);
expect(job.result.status).toBe(400); expect(job.result.status).toBe(400);
}); });
@ -222,7 +193,7 @@ describe('workflow > instructions > request', () => {
const n1 = await workflow.createNode({ const n1 = await workflow.createNode({
type: 'request', type: 'request',
config: { config: {
url: testUrl, url: URL_DATA,
method: 'POST', method: 'POST',
data: '{"title": "<%=ctx.data.title%>"}', data: '{"title": "<%=ctx.data.title%>"}',
} as RequestConfig, } as RequestConfig,
@ -234,14 +205,8 @@ describe('workflow > instructions > request', () => {
let [execution] = await workflow.getExecutions(); let [execution] = await workflow.getExecutions();
let [job] = await execution.getJobs(); let [job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.PENDING);
await sleep(1000);
[execution] = await workflow.getExecutions();
[job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.RESOLVED); expect(job.status).toEqual(JOB_STATUS.RESOLVED);
expect(JSON.parse(job.result.data)).toEqual({ title: 't1' }); expect(job.result.data).toEqual({ title: 't1' });
}); });
}); });
}); });

View File

@ -1,12 +1,13 @@
import { JOB_STATUS } from '../constants';
import { render } from 'ejs'; import { render } from 'ejs';
import axios, { AxiosRequestConfig } from 'axios'; import axios, { AxiosRequestConfig } from 'axios';
import _ from 'lodash'; import _ from 'lodash';
import { Instruction } from './index'; import { Instruction } from './index';
import { JOB_STATUS } from '../constants';
export interface Header { export interface Header {
name: string; name: string;
value: any; value: string;
} }
export type RequestConfig = Pick<AxiosRequestConfig, 'url' | 'method' | 'data' | 'timeout'> & { export type RequestConfig = Pick<AxiosRequestConfig, 'url' | 'method' | 'data' | 'timeout'> & {
@ -14,24 +15,15 @@ export type RequestConfig = Pick<AxiosRequestConfig, 'url' | 'method' | 'data' |
ignoreFail: boolean; ignoreFail: boolean;
}; };
async function renderData(templateStr: string, vars: any): Promise<string> {
if (_.isEmpty(templateStr)) {
return '';
}
return render(templateStr, vars, { async: true });
}
async function triggerResume(plugin, job, status, result) {
job.set('status', status);
job.set('result', result);
return plugin.resume(job);
}
export default class implements Instruction { export default class implements Instruction {
async run(node, input, processor) { constructor(public plugin) {}
request = async (node, job, processor) => {
const templateVars = { const templateVars = {
node: processor.jobsMapByNodeId, node: processor.jobsMapByNodeId,
ctx: processor.execution.context, ctx: processor.execution.context,
$jobsMapByNodeId: processor.jobsMapByNodeId,
$context: processor.execution.context,
}; };
const requestConfig = node.config as RequestConfig; const requestConfig = node.config as RequestConfig;
@ -39,48 +31,54 @@ export default class implements Instruction {
const headers = { const headers = {
'Content-Type': 'application/json', 'Content-Type': 'application/json',
}; };
const { headers: headerArr = [], ignoreFail = false, method = 'POST', timeout = 5000 } = requestConfig; const { headers: headerArr = [], method = 'POST', timeout = 5000 } = requestConfig;
headerArr.forEach((header) => (headers[header.name] = header.value)); headerArr.forEach((header) => (headers[header.name] = header.value));
let url, data; let url, data;
try { try {
url = await renderData(requestConfig.url, templateVars); url = await render(requestConfig.url!.trim(), templateVars, { async: true });
} catch (e) { data = requestConfig.data ? await render(requestConfig.data.trim(), templateVars, { async: true }) : undefined;
console.warn(e);
throw new Error(`ejs can't render url, please check url format${(e as Error)?.message}`); } catch (error2) {
} // console.error(error2);
try { job.set({
data = await renderData(requestConfig.data, templateVars); status: JOB_STATUS.REJECTED,
} catch (e) { result: error2.message
console.warn(e); });
throw new Error(`ejs can't render request data, please check request data format${(e as Error)?.message}`);
} }
try {
const response = await axios.request({
url,
method,
headers,
data,
timeout,
});
job.set({
status: JOB_STATUS.RESOLVED,
result: response.data
});
} catch (error1) {
// console.error('axios error?', error1);
job.set({
status: JOB_STATUS.REJECTED,
result: error1.isAxiosError ? error1.toJSON() : error1.message
});
}
return this.plugin.resume(job);
};
async run(node, input, processor) {
const job = await processor.saveJob({ const job = await processor.saveJob({
status: JOB_STATUS.PENDING, status: JOB_STATUS.PENDING,
nodeId: node.id, nodeId: node.id,
}); });
const plugin = processor.options.plugin; setTimeout(() => {
axios this.request(node, job, processor);
.request({ });
method,
timeout,
headers,
url,
data,
})
.then((resp) => {
if (resp.status >= 200 && resp.status < 300) {
triggerResume(plugin, job, JOB_STATUS.RESOLVED, resp.data);
} else {
triggerResume(plugin, job, JOB_STATUS.REJECTED, resp);
}
})
.catch((e) => {
console.warn(e);
triggerResume(plugin, job, JOB_STATUS.REJECTED, (e as Error)?.message);
});
return job; return job;
} }