| 539 | } |
| 540 | |
| 541 | _runJob(job) { |
| 542 | let psTimeout = null, |
| 543 | completed = false; |
| 544 | |
| 545 | const preventStalling = () => { |
| 546 | psTimeout = null; |
| 547 | if (this._isClosed) return; |
| 548 | finally_(this._preventStall(job.id), () => { |
| 549 | if (completed || this._isClosed) return; |
| 550 | const interval = this.settings.stallInterval / 2; |
| 551 | psTimeout = setTimeout(preventStalling, interval); |
| 552 | }).catch(this._emitErrorAfterTick); |
| 553 | }; |
| 554 | preventStalling(); |
| 555 | |
| 556 | const handleOutcome = (err, data) => { |
| 557 | completed = true; |
| 558 | if (psTimeout) { |
| 559 | clearTimeout(psTimeout); |
| 560 | psTimeout = null; |
| 561 | } |
| 562 | |
| 563 | return this._finishJob(err, data, job); |
| 564 | }; |
| 565 | |
| 566 | let promise = this.handler(job); |
| 567 | |
| 568 | if (job.options.timeout) { |
| 569 | const message = `Job ${job.id} timed out (${job.options.timeout} ms)`; |
| 570 | promise = helpers.withTimeout(promise, job.options.timeout, message); |
| 571 | } |
| 572 | |
| 573 | const jobPromise = finally_( |
| 574 | promise.then((data) => handleOutcome(null, data), handleOutcome), |
| 575 | // The only error that can happen here is either network- or |
| 576 | // Redis-related, or if Queue#close times out while a job is processing, |
| 577 | // and the job later finishes. |
| 578 | () => this.activeJobs.delete(jobPromise) |
| 579 | ); |
| 580 | |
| 581 | // We specifically use the value produced by then to avoid cases where the |
| 582 | // process handler returns the same Promise object each invocation. |
| 583 | this.activeJobs.add(jobPromise); |
| 584 | return jobPromise; |
| 585 | } |
| 586 | |
| 587 | _preventStall(jobId) { |
| 588 | return helpers.callAsync((done) => |