tachybase_todo/packages/plugins/@nocobase/plugin-workflow-request/src/server/RequestInstruction.ts
Junyi 4b8915b616
refactor(plugin-workflow): add sync option for trigger (#3383)
* refactor(plugin-workflow): add sync option for trigger

* feat(plugin-workflow-request): support sync call in request

* fix(plugin-workflow-request): fix request async call

* refactor(plugin-workflow): add method to check if workflow is sync

* fix(plugin-workflow): fix useAvailable in nodes

* fix(plugin-workflow): fix node.isAvailable check

* test(plugin-workflow): limit mysql version to pass test

* fix(plugin-workflow-delay): fix test case

* fix(plugin-workflow-delay): fix test case

* feat(plugin-workflow): add sync field for workflow

* refactor(plugin-workflow): adjust end node logic

* fix(plugin-workflow): support sync mode in form trigger

* feat(plugin-workflow): add end instruction

* fix(plugin-workflow-form-trigger): fix test cases

* fix(plugin-workflow): fix transaction for sync event
2024-01-25 22:10:03 +08:00

103 lines
2.8 KiB
TypeScript

import axios, { AxiosRequestConfig } from 'axios';
import { Processor, Instruction, JOB_STATUS, FlowNodeModel } from '@nocobase/plugin-workflow';
export interface Header {
name: string;
value: string;
}
export type RequestConfig = Pick<AxiosRequestConfig, 'url' | 'method' | 'params' | 'data' | 'timeout'> & {
headers: Array<Header>;
ignoreFail: boolean;
};
async function request(config) {
// default headers
const { url, method = 'POST', data, timeout = 5000 } = config;
const headers = (config.headers ?? []).reduce((result, header) => {
if (header.name.toLowerCase() === 'content-type') {
return result;
}
return Object.assign(result, { [header.name]: header.value });
}, {});
const params = (config.params ?? []).reduce(
(result, param) => Object.assign(result, { [param.name]: param.value }),
{},
);
// TODO(feat): only support JSON type for now, should support others in future
headers['Content-Type'] = 'application/json';
return axios.request({
url,
method,
headers,
params,
data,
timeout,
});
}
export default class extends Instruction {
async run(node: FlowNodeModel, prevJob, processor: Processor) {
const config = processor.getParsedValue(node.config, node.id) as RequestConfig;
const { workflow } = processor.execution;
const sync = this.workflow.isWorkflowSync(workflow);
if (sync) {
try {
const response = await request(config);
return {
status: JOB_STATUS.RESOLVED,
result: response.data,
};
} catch (error) {
return {
status: JOB_STATUS.FAILED,
result: error.isAxiosError ? error.toJSON() : error.message,
};
}
}
const job = await processor.saveJob({
status: JOB_STATUS.PENDING,
nodeId: node.id,
nodeKey: node.key,
upstreamId: prevJob?.id ?? null,
});
// eslint-disable-next-line promise/catch-or-return
request(config)
.then((response) => {
job.set({
status: JOB_STATUS.RESOLVED,
result: response.data,
});
})
.catch((error) => {
job.set({
status: JOB_STATUS.FAILED,
result: error.isAxiosError ? error.toJSON() : error.message,
});
})
.finally(() => {
processor.logger.info(`request (#${node.id}) response received, status: ${job.get('status')}`);
this.workflow.resume(job);
});
processor.logger.info(`request (#${node.id}) sent to "${config.url}", waiting for response...`);
return processor.exit();
}
async resume(node: FlowNodeModel, job, processor: Processor) {
const { ignoreFail } = node.config as RequestConfig;
if (ignoreFail) {
job.set('status', JOB_STATUS.RESOLVED);
}
return job;
}
}