From a1127300aefcfbd9e0968b6b8c4ccf16a65268dc Mon Sep 17 00:00:00 2001 From: Junyi Date: Thu, 12 Jan 2023 18:22:06 +0800 Subject: [PATCH] refactor(plugin-workflow): refactor request instruction (#1356) --- packages/plugins/workflow/package.json | 1 + .../plugins/workflow/src/server/Processor.ts | 12 +- .../__tests__/instructions/request.test.ts | 139 +++++++----------- .../src/server/instructions/request.ts | 92 ++++++------ 4 files changed, 104 insertions(+), 140 deletions(-) diff --git a/packages/plugins/workflow/package.json b/packages/plugins/workflow/package.json index 198c00344..c46ef4dc5 100644 --- a/packages/plugins/workflow/package.json +++ b/packages/plugins/workflow/package.json @@ -12,6 +12,7 @@ "@nocobase/server": "0.9.0-alpha.1", "@nocobase/utils": "0.9.0-alpha.1", "antd": "4.22.8", + "axios": "^0.27.2", "classnames": "^2.3.1", "cron-parser": "4.4.0", "ejs": "^3.1.8", diff --git a/packages/plugins/workflow/src/server/Processor.ts b/packages/plugins/workflow/src/server/Processor.ts index 725340125..08873c72b 100644 --- a/packages/plugins/workflow/src/server/Processor.ts +++ b/packages/plugins/workflow/src/server/Processor.ts @@ -176,12 +176,12 @@ export default class Processor { public async run(node, input?) { const { instructions } = this.options.plugin; - const { run } = instructions.get(node.type); - if (typeof run !== 'function') { + const instruction = instructions.get(node.type); + if (typeof instruction.run !== 'function') { 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 @@ -200,12 +200,12 @@ export default class Processor { async recall(node, job) { const { instructions } = this.options.plugin; - const { resume } = instructions.get(node.type); - if (typeof resume !== 'function') { + const instruction = instructions.get(node.type); + if (typeof instruction.resume !== 'function') { 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) { diff --git a/packages/plugins/workflow/src/server/__tests__/instructions/request.test.ts b/packages/plugins/workflow/src/server/__tests__/instructions/request.test.ts index 325a62948..b86853fe1 100644 --- a/packages/plugins/workflow/src/server/__tests__/instructions/request.test.ts +++ b/packages/plugins/workflow/src/server/__tests__/instructions/request.test.ts @@ -1,37 +1,14 @@ -/** - * @jest-environment node - */ - import { Application } from '@nocobase/server'; import Database from '@nocobase/database'; import { getApp, sleep } from '..'; import { RequestConfig } from '../../instructions/request'; -import axios, { AxiosRequestConfig } from 'axios'; import { JOB_STATUS } from '../../constants'; -const testUrl = 'https://nocobase.com/test'; -const url_400 = 'https://nocobase.com/400'; -const timeoutUrl = 'https://nocobase.com/timeout'; -jest.mock('axios', () => { - return { - 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, - }; - }, - }; -}); +const PORT = 12345; + +const URL_DATA = `http://localhost:${PORT}/data`; +const URL_400 = `http://localhost:${PORT}/api/400`; +const URL_TIMEOUT = `http://localhost:${PORT}/timeout`; describe('workflow > instructions > request', () => { let app: Application; @@ -41,7 +18,27 @@ describe('workflow > instructions > request', () => { let workflow; 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; WorkflowModel = db.getCollection('workflows').model; @@ -65,7 +62,7 @@ describe('workflow > instructions > request', () => { await workflow.createNode({ type: 'request', config: { - url: 'https://www.baidu.com?name=lily&age=20', + url: URL_DATA, method: 'GET', } as RequestConfig, }); @@ -76,12 +73,6 @@ describe('workflow > instructions > request', () => { let [execution] = await workflow.getExecutions(); 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); }); @@ -89,7 +80,7 @@ describe('workflow > instructions > request', () => { await workflow.createNode({ type: 'request', config: { - url: 'https://www.xxx.com?title=<%= ctx.data.title // 测试 %>', + url: `${URL_DATA}?title=<%= ctx.data.title %>`, method: 'GET', } as RequestConfig, }); @@ -100,77 +91,69 @@ describe('workflow > instructions > request', () => { let [execution] = await workflow.getExecutions(); 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.result.url).toEqual('https://www.xxx.com?title=t1'); + console.log(job.result); + expect(job.result.meta.title).toEqual('t1'); }); it('request - timeout', async () => { await workflow.createNode({ type: 'request', config: { - url: timeoutUrl, + url: URL_TIMEOUT, method: 'GET', - timeout: 1000, + timeout: 250, } as RequestConfig, }); await PostRepo.create({ values: { title: 't1' } }); - await sleep(500); + await sleep(1000); let [execution] = await workflow.getExecutions(); 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.result).toMatch('timeout'); + expect(job.result).toMatchObject({ + code: 'ECONNABORTED', + name: 'AxiosError', + status: null, + message: 'timeout of 250ms exceeded' + }); }); it('request - ignoreFail', async () => { await workflow.createNode({ type: 'request', config: { - url: timeoutUrl, + url: URL_TIMEOUT, method: 'GET', - timeout: 1000, + timeout: 250, ignoreFail: true, } as RequestConfig, }); await PostRepo.create({ values: { title: 't1' } }); - await sleep(500); + await sleep(1000); let [execution] = await workflow.getExecutions(); 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.result).toMatch('timeout'); + expect(job.result).toMatchObject({ + code: 'ECONNABORTED', + name: 'AxiosError', + status: null, + message: 'timeout of 250ms exceeded' + }); }); it('response 400', async () => { await workflow.createNode({ type: 'request', config: { - url: url_400, + url: URL_400, method: 'GET', - timeout: 1000, ignoreFail: false, } as RequestConfig, }); @@ -181,12 +164,6 @@ describe('workflow > instructions > request', () => { let [execution] = await workflow.getExecutions(); 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.result.status).toBe(400); }); @@ -195,7 +172,7 @@ describe('workflow > instructions > request', () => { await workflow.createNode({ type: 'request', config: { - url: url_400, + url: URL_400, method: 'GET', timeout: 1000, ignoreFail: true, @@ -208,12 +185,6 @@ describe('workflow > instructions > request', () => { let [execution] = await workflow.getExecutions(); 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.result.status).toBe(400); }); @@ -222,7 +193,7 @@ describe('workflow > instructions > request', () => { const n1 = await workflow.createNode({ type: 'request', config: { - url: testUrl, + url: URL_DATA, method: 'POST', data: '{"title": "<%=ctx.data.title%>"}', } as RequestConfig, @@ -234,14 +205,8 @@ describe('workflow > instructions > request', () => { let [execution] = await workflow.getExecutions(); 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(JSON.parse(job.result.data)).toEqual({ title: 't1' }); + expect(job.result.data).toEqual({ title: 't1' }); }); }); }); diff --git a/packages/plugins/workflow/src/server/instructions/request.ts b/packages/plugins/workflow/src/server/instructions/request.ts index 54b4448d6..b71ccbc20 100644 --- a/packages/plugins/workflow/src/server/instructions/request.ts +++ b/packages/plugins/workflow/src/server/instructions/request.ts @@ -1,12 +1,13 @@ -import { JOB_STATUS } from '../constants'; import { render } from 'ejs'; import axios, { AxiosRequestConfig } from 'axios'; import _ from 'lodash'; + import { Instruction } from './index'; +import { JOB_STATUS } from '../constants'; export interface Header { name: string; - value: any; + value: string; } export type RequestConfig = Pick & { @@ -14,24 +15,15 @@ export type RequestConfig = Pick { - 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 { - async run(node, input, processor) { + constructor(public plugin) {} + + request = async (node, job, processor) => { const templateVars = { node: processor.jobsMapByNodeId, ctx: processor.execution.context, + $jobsMapByNodeId: processor.jobsMapByNodeId, + $context: processor.execution.context, }; const requestConfig = node.config as RequestConfig; @@ -39,48 +31,54 @@ export default class implements Instruction { const headers = { '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)); let url, data; try { - url = await renderData(requestConfig.url, templateVars); - } catch (e) { - console.warn(e); - throw new Error(`ejs can't render url, please check url format!${(e as Error)?.message}`); - } - try { - data = await renderData(requestConfig.data, templateVars); - } catch (e) { - console.warn(e); - throw new Error(`ejs can't render request data, please check request data format!${(e as Error)?.message}`); + url = await render(requestConfig.url!.trim(), templateVars, { async: true }); + data = requestConfig.data ? await render(requestConfig.data.trim(), templateVars, { async: true }) : undefined; + + } catch (error2) { + // console.error(error2); + job.set({ + status: JOB_STATUS.REJECTED, + result: error2.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({ status: JOB_STATUS.PENDING, nodeId: node.id, }); - const plugin = processor.options.plugin; - axios - .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); - }); + setTimeout(() => { + this.request(node, job, processor); + }); return job; }