mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2025-02-02 07:01:30 -08:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor(core): Suppress MaxListenersExceededWarning in the logs (#10077)

Browse source
This commit is contained in:
कारतोफ्फेलस्क्रिप्ट™ 2024-07-16 18:20:55 +02:00 committed by GitHub
parent 48f047ee2e
commit 3bbeae47f3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 22 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
packages/cli/src/Queue.ts
View file

@ -35,6 +35,24 @@ export interface WebhookResponse {
export class Queue { export class Queue {
private jobQueue: JobQueue; private jobQueue: JobQueue;
/**
* The number of jobs a single server can process concurrently
* Any worker that wants to process executions must first set this to a non-zero value
*/
private concurrency = 0;
setConcurrency(concurrency: number) {
this.concurrency = concurrency;
// This sets the max event listeners on the jobQueue EventEmitter to prevent the logs getting flooded with MaxListenersExceededWarning
// see: https://github.com/OptimalBits/bull/blob/develop/lib/job.js#L497-L521
this.jobQueue.setMaxListeners(
4 + // `close`
2 + // `error`
2 + // `global:progress`
concurrency * 2, // 2 global events for every call to `job.finished()`
);
}
constructor(private activeExecutions: ActiveExecutions) {} constructor(private activeExecutions: ActiveExecutions) {}
async init() { async init() {
@ -102,8 +120,8 @@ export class Queue {
return new Set(inProgressJobs.map((job) => job.data.executionId)); return new Set(inProgressJobs.map((job) => job.data.executionId));
} }
async process(concurrency: number, fn: Bull.ProcessCallbackFunction<JobData>): Promise<void> { async process(fn: Bull.ProcessCallbackFunction<JobData>): Promise<void> {
return await this.jobQueue.process(concurrency, fn); return await this.jobQueue.process(this.concurrency, fn);
} }
async ping(): Promise<string> { async ping(): Promise<string> {

6
packages/cli/src/commands/worker.ts
View file

@ -320,11 +320,9 @@ export class Worker extends BaseCommand {
const envConcurrency = config.getEnv('executions.concurrency.productionLimit'); const envConcurrency = config.getEnv('executions.concurrency.productionLimit');
const concurrency = envConcurrency !== -1 ? envConcurrency : flags.concurrency; const concurrency = envConcurrency !== -1 ? envConcurrency : flags.concurrency;
Worker.jobQueue.setConcurrency(concurrency);
void Worker.jobQueue.process( void Worker.jobQueue.process(async (job) => await this.runJob(job, this.nodeTypes));
concurrency,
async (job) => await this.runJob(job, this.nodeTypes),
);
Worker.jobQueue.getBullObjectInstance().on('global:progress', (jobId: JobId, progress) => { Worker.jobQueue.getBullObjectInstance().on('global:progress', (jobId: JobId, progress) => {
// Progress of a job got updated which does get used // Progress of a job got updated which does get used