Skip to content

Commit

Permalink
fix: tests, agenda-instance should have a smaller processEvery
Browse files Browse the repository at this point in the history
  • Loading branch information
simllll committed Oct 14, 2020
1 parent 71ff8a2 commit b248a2b
Show file tree
Hide file tree
Showing 6 changed files with 45 additions and 32 deletions.
2 changes: 1 addition & 1 deletion .mocharc.jsonc
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
"recursive": true,
"reporter": "spec",
"slow": 75,
"timeout": 5000,
"timeout": 25000,
"ui": "bdd",
"exit": true
}
8 changes: 4 additions & 4 deletions package.json
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,10 @@
"build": "tsc",
"test": "npm run lint && npm run mocha",
"lint": "eslint",
"mocha": "mocha -r ts-node/register --reporter spec --timeout 8000 -b",
"mocha-debug": "DEBUG=agenda:**,-agenda:internal:** mocha -r ts-node/register --reporter spec --timeout 8000 -b",
"mocha-debug-internal": "DEBUG=agenda:internal:** mocha -r ts-node/register --reporter spec --timeout 8000 -b",
"mocha-debug-all": "DEBUG=agenda:** mocha -r ts-node/register --reporter spec --timeout 8000 -b",
"mocha": "mocha -r ts-node/register --reporter spec -b",
"mocha-debug": "DEBUG=agenda:**,-agenda:internal:** mocha -r ts-node/register --reporter spec -b",
"mocha-debug-internal": "DEBUG=agenda:internal:** mocha -r ts-node/register --reporter spec -b",
"mocha-debug-all": "DEBUG=agenda:** mocha -r ts-node/register --reporter spec -b",
"docs": "jsdoc --configure .jsdoc.json --verbose"
},
"config": {
Expand Down
59 changes: 35 additions & 24 deletions src/JobProcessor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,7 @@ export class JobProcessor {
) {
log('creating interval to call processJobs every [%dms]', processEvery);
this.processInterval = setInterval(() => this.process(), processEvery);
this.process();
}

stop(): Job[] {
Expand Down Expand Up @@ -345,34 +346,33 @@ export class JobProcessor {
// Check if there is any job that is not blocked by concurrency
const job = this.jobQueue.returnNextConcurrencyFreeJob(this.jobStatus);

if (job) {
if (!job) {
log.extend('jobProcessing')('[%s:%s] there is no job to process');
return;
}

log.extend('jobProcessing')('[%s:%s] there is a job to process', job.attrs.name, job.attrs._id);

// If the 'nextRunAt' time is older than the current time, run the job
// Otherwise, setTimeout that gets called at the time of 'nextRunAt'
if (job.attrs.nextRunAt <= now) {
log.extend('jobProcessing')(
'[%s:%s] there is a job to process',
'[%s:%s] nextRunAt is in the past, run the job immediately',
job.attrs.name,
job.attrs._id
);

// If the 'nextRunAt' time is older than the current time, run the job
// Otherwise, setTimeout that gets called at the time of 'nextRunAt'
if (job.attrs.nextRunAt <= now) {
log.extend('jobProcessing')(
'[%s:%s] nextRunAt is in the past, run the job immediately',
job.attrs.name,
job.attrs._id
);
this.runOrRetry(job);
} else {
const runIn = job.attrs.nextRunAt.getTime() - now.getTime();
log.extend('jobProcessing')(
'[%s:%s] nextRunAt is in the future, calling setTimeout(%d)',
job.attrs.name,
job.attrs._id,
runIn
);
setTimeout(() => {
this.jobProcessing();
}, runIn);
}
this.runOrRetry(job);
} else {
const runIn = job.attrs.nextRunAt.getTime() - now.getTime();
log.extend('jobProcessing')(
'[%s:%s] nextRunAt is in the future, calling setTimeout(%d)',
job.attrs.name,
job.attrs._id,
runIn
);
setTimeout(() => {
this.jobProcessing();
}, runIn);
}
}

Expand Down Expand Up @@ -437,6 +437,11 @@ export class JobProcessor {
log.extend('runOrRetry')('[%s:%s] processing job', job.attrs.name, job.attrs._id);
// CALL THE ACTUAL METHOD TO PROCESS THE JOB!!!
await job.run();
log.extend('runOrRetry')(
'[%s:%s] processing job successfull',
job.attrs.name,
job.attrs._id
);

// Job isn't in running jobs so throw an error
if (!this.runningJobs.includes(job)) {
Expand All @@ -449,6 +454,12 @@ export class JobProcessor {
);
}
} catch (err) {
log.extend('runOrRetry')(
'[%s:%s] processing job failed',
job.attrs.name,
job.attrs._id,
err
);
job.agenda.emit('error', err);
} finally {
// Remove the job from the running queue
Expand Down
3 changes: 2 additions & 1 deletion src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,8 @@ export class Agenda extends EventEmitter {

this.attrs = {
name: config.name || '',
processEvery: humanInterval(config.processEvery) || humanInterval('5 seconds'),
processEvery:
(config.processEvery && humanInterval(config.processEvery)) || humanInterval('5 seconds'),
defaultConcurrency: config.defaultConcurrency || 5,
maxConcurrency: config.maxConcurrency || 20,
defaultLockLimit: config.defaultLockLimit || 0,
Expand Down
3 changes: 2 additions & 1 deletion test/fixtures/agenda-instance.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,8 @@ const agenda = new Agenda(
{
db: {
address: connStr
}
},
processEvery: 100
},
async () => {
tests.forEach(test => {
Expand Down
2 changes: 1 addition & 1 deletion test/job.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1203,7 +1203,7 @@ describe('Job', () => {
});

let ran1 = false;
let ran2 = true;
let ran2 = false;
let doneCalled = false;

const serviceError = function (e) {
Expand Down

0 comments on commit b248a2b

Please sign in to comment.