feat(agenda): add drainJobsBeforeClose option to agenda settings

tsedio · Jan 3, 2024 · 3c10b05 · 3c10b05
1 parent 8c8472f
commit 3c10b05
Show file tree

Hide file tree

Showing 6 changed files with 54 additions and 7 deletions.
diff --git a/docs/tutorials/agenda.md b/docs/tutorials/agenda.md
@@ -40,6 +40,8 @@ const mongoConnectionString = "mongodb://127.0.0.1/agenda";
 @Configuration({
   agenda: {
     enabled: true, // Enable Agenda jobs for this instance.
+    // drainJobsBeforeStop: true, // Wait for jobs to finish before stopping the agenda process.
+    // disableJobProcessing: true, // Prevents jobs from being processed.
     // pass any options that you would normally pass to new Agenda(), e.g.
     db: {
       address: mongoConnectionString

diff --git a/packages/third-parties/agenda/jest.config.js b/packages/third-parties/agenda/jest.config.js
@@ -7,7 +7,7 @@ module.exports = {
   coverageThreshold: {
     global: {
       statements: 100,
-      branches: 91.48,
+      branches: 91.66,
       functions: 100,
       lines: 100
     }

diff --git a/packages/third-parties/agenda/readme.md b/packages/third-parties/agenda/readme.md
@@ -61,6 +61,8 @@ const mongoConnectionString = "mongodb://127.0.0.1/agenda";
 @Configuration({
   agenda: {
     enabled: true, // Enable Agenda jobs for this instance.
+    // drainJobsBeforeStop: true, // Wait for jobs to finish before stopping the agenda process.
+    // disableJobProcessing: true, // Prevents jobs from being processed.
     // pass any options that you would normally pass to new Agenda(), e.g.
     db: {
       address: mongoConnectionString

diff --git a/packages/third-parties/agenda/src/AgendaModule.spec.ts b/packages/third-parties/agenda/src/AgendaModule.spec.ts
@@ -11,6 +11,7 @@ jest.mock("agenda", () => {
     Agenda: class {
       close = jest.fn();
       stop = jest.fn();
+      drain = jest.fn();
       define = jest.fn();
       every = jest.fn();
       schedule = jest.fn();
@@ -131,11 +132,32 @@ describe("AgendaModule", () => {
 
         await agendaModule.$onDestroy();
 
+        expect(agendaModule.agenda.stop).toHaveBeenCalledWith();
         expect(agendaModule.agenda.close).toHaveBeenCalledWith({force: true});
       });
     });
   });
+  describe("when agenda is enabled and drainJobsBeforeClose = true", () => {
+    beforeEach(() =>
+      PlatformTest.create({
+        agenda: {
+          enabled: true,
+          drainJobsBeforeClose: true
+        }
+      })
+    );
+    afterEach(() => PlatformTest.reset());
+    describe("$onDestroy()", () => {
+      it("should close agenda", async () => {
+        const agendaModule = PlatformTest.get<any>(AgendaModule)!;
+
+        await agendaModule.$onDestroy();
 
+        expect(agendaModule.agenda.drain).toHaveBeenCalledWith();
+        expect(agendaModule.agenda.close).toHaveBeenCalledWith({force: true});
+      });
+    });
+  });
   describe("when agenda is enabled but disableJobProcessing = true", () => {
     beforeEach(() =>
       PlatformTest.create({

diff --git a/packages/third-parties/agenda/src/AgendaModule.ts b/packages/third-parties/agenda/src/AgendaModule.ts
@@ -20,6 +20,9 @@ export class AgendaModule implements OnDestroy, AfterListen {
   @Constant("agenda.enabled", false)
   private enabled: boolean;
 
+  @Constant("agenda.drainJobsBeforeClose", false)
+  private drainJobsBeforeClose: boolean;
+
   @Constant("agenda.disableJobProcessing", false)
   private disableJobProcessing: boolean;
 
@@ -28,7 +31,10 @@ export class AgendaModule implements OnDestroy, AfterListen {
       const providers = this.getProviders();
 
       if (!this.disableJobProcessing) {
-        this.logger.info("Agenda add definitions...");
+        this.logger.info({
+          event: "AGENDA_ADD_DEFINITIONS",
+          message: "Agenda add definitions"
+        });
         providers.forEach((provider) => this.addAgendaDefinitionsForProvider(provider));
 
         await this.injector.emit("$beforeAgendaStart");
@@ -37,22 +43,36 @@ export class AgendaModule implements OnDestroy, AfterListen {
       await this.agenda.start();
 
       if (!this.disableJobProcessing) {
-        this.logger.info("Agenda add scheduled jobs...");
+        this.logger.info({
+          event: "AGENDA_ADD_JOBS",
+          message: "Agenda add scheduled jobs"
+        });
         await Promise.all(providers.map((provider) => this.scheduleJobsForProvider(provider)));
 
         await this.injector.emit("$afterAgendaStart");
       }
-    } else {
-      this.logger.info("Agenda disabled...");
     }
   }
 
   async $onDestroy(): Promise<any> {
     if (this.enabled) {
-      await this.agenda.stop();
+      if (this.drainJobsBeforeClose) {
+        this.logger.info({
+          event: "AGENDA_DRAIN",
+          message: "Agenda is draining all jobs before close"
+        });
+        await this.agenda.drain();
+      } else {
+        this.logger.info({
+          event: "AGENDA_STOP",
+          message: "Agenda is stopping"
+        });
+        await this.agenda.stop();
+      }
+
       await this.agenda.close({force: true});
 
-      this.logger.info("Agenda stopped...");
+      this.logger.info({event: "AGENDA_STOP", message: "Agenda stopped"});
     }
   }
 

diff --git a/packages/third-parties/agenda/src/interfaces/interfaces.ts b/packages/third-parties/agenda/src/interfaces/interfaces.ts
@@ -9,6 +9,7 @@ declare global {
          */
         enabled?: boolean;
         disableJobProcessing?: boolean;
+        drainJobsBeforeClose?: boolean;
       } & AgendaConfig;
     }
   }