Skip to content

Regression: Allow apps to schedule jobs along with processor register #19416

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sampaiodiego merged 14 commits into from
Nov 9, 2020
Merged

Regression: Allow apps to schedule jobs along with processor register #19416

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
f4bcd67
Make the apps scheduler run jobs after defining them
thassiov Oct 29, 2020
041afe2
Ensure onetime jobs won't have duplicates at startup
thassiov Oct 30, 2020
48eb5bc
Make the call to `registerProcessors` sync
thassiov Oct 30, 2020
b81fece
Revert "Make the call to `registerProcessors` sync"
thassiov Nov 3, 2020
1bbb231
Receive data from processors' `startupSetting`
thassiov Nov 4, 2020
f0e81d6
Make only the `data` object available from the `job` object
thassiov Nov 4, 2020
bd07875
Merge branch 'develop' into impr/run-job-after-register
d-gubert Nov 4, 2020
f55feb1
Update Apps-Engine
d-gubert Nov 4, 2020
efb967a
Merge remote-tracking branch 'origin/develop' into impr/run-job-after…
d-gubert Nov 6, 2020
813502e
Change collection name for apps scheduler
d-gubert Nov 6, 2020
3cad659
Update app/apps/server/bridges/scheduler.js
d-gubert Nov 9, 2020
13cce23
Rename recurring job's prop from `cron` to `interval`
thassiov Nov 9, 2020
7fe3ab5
Update Apps-Engine
d-gubert Nov 9, 2020
11f9e54
Merge branch 'develop' into impr/run-job-after-register
d-gubert Nov 9, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 35 additions & 5 deletions app/apps/server/bridges/scheduler.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,23 +1,46 @@
import Agenda from 'agenda';
import { MongoInternals } from 'meteor/mongo';
import { StartupType } from '@rocket.chat/apps-engine/definition/scheduler';

function _callProcessor(processor) {
return (job) => processor(job.attrs.data);
return (job) => processor(job?.attrs?.data || {});
}

export class AppSchedulerBridge {
constructor(orch) {
this.orch = orch;
this.scheduler = new Agenda({
mongo: MongoInternals.defaultRemoteCollectionDriver().mongo.client.db(),
collection: 'rocketchat_agenda_jobs',
db: { collection: 'rocketchat_apps_scheduler' },
// this ensures the same job doesn't get executed multiple times in a cluster
defaultConcurrency: 1,
});
this.isConnected = false;
}

async registerProcessors(processors = [], appId) {
const runAfterRegister = [];
this.orch.debugLog(`The App ${ appId } is registering job processors`, processors);
processors.forEach(({ id, processor }) => this.scheduler.define(id, _callProcessor(processor)));
processors.forEach(({ id, processor, startupSetting }) => {
this.scheduler.define(id, _callProcessor(processor));

if (startupSetting) {
switch (startupSetting.type) {
case StartupType.ONETIME:
runAfterRegister.push(this.scheduleOnceAfterRegister({ id, when: startupSetting.when, data: startupSetting.data }, appId));
break;
case StartupType.RECURRING:
runAfterRegister.push(this.scheduleRecurring({ id, interval: startupSetting.interval, data: startupSetting.data }, appId));
break;
default:
break;
}
}
});

if (runAfterRegister.length) {
await Promise.all(runAfterRegister);
}
}

async scheduleOnce(job, appId) {
Expand All @@ -26,10 +49,17 @@ export class AppSchedulerBridge {
await this.scheduler.schedule(job.when, job.id, job.data || {});
}

async scheduleOnceAfterRegister(job, appId) {
const scheduledJobs = await this.scheduler.jobs({ name: job.id, type: 'normal' });
if (!scheduledJobs.length) {
await this.scheduleOnce(job, appId);
}
}

async scheduleRecurring(job, appId) {
this.orch.debugLog(`The App ${ appId } is scheduling a recurring job`, job);
await this.startScheduler();
await this.scheduler.every(job.cron, job.id, job.data || {});
await this.scheduler.every(job.interval, job.id, job.data || {});
}

async cancelJob(jobId, appId) {
Expand All @@ -45,7 +75,7 @@ export class AppSchedulerBridge {
async cancelAllJobs(appId) {
this.orch.debugLog(`Canceling all jobs of App ${ appId }`);
await this.startScheduler();
const matcher = new RegExp(`^_${ appId }`);
const matcher = new RegExp(`_${ appId }$`);
try {
await this.scheduler.cancel({ name: { $regex: matcher } });
} catch (e) {
Expand Down
19 changes: 6 additions & 13 deletions package-lock.json
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@
"@nivo/heatmap": "^0.61.0",
"@nivo/line": "^0.61.1",
"@nivo/pie": "^0.61.1",
"@rocket.chat/apps-engine": "1.19.0-beta.4014",
"@rocket.chat/apps-engine": "1.19.0-beta.4086",
"@rocket.chat/css-in-js": "^0.17.2",
"@rocket.chat/emitter": "^0.17.2",
"@rocket.chat/fuselage": "^0.17.2",
Expand Down