Queue now respects max items.

Author: niemyjski

packages/core/src/queue/DefaultEventQueue.ts

@@ -50,7 +50,6 @@ export class DefaultEventQueue implements IEventQueue {
   private _queue: EventQueueItem[] = [];
   private _loadPersistedEvents = true;
 
-  // TODO: Implement support for max queue items.
   constructor(
     private config: Configuration,
     private maxItems: number = 250
@@ -277,12 +276,21 @@ export class DefaultEventQueue implements IEventQueue {
     this._lastFileTimestamp = Math.max(Date.now(), this._lastFileTimestamp + 1);
     const file = `${this.QUEUE_PREFIX}${this._lastFileTimestamp}.json`;
 
-    this._queue.push({ file, event });
-    if (this.config.usePersistedQueueStorage) {
+    const { storage, log } = this.config.services;
+    const useStorage: boolean = this.config.usePersistedQueueStorage;
+    if (this._queue.push({ file, event }) > this.maxItems) {
+      log.trace("Removing oldest queue entry: maxItems exceeded");
+      const item = this._queue.shift();
+      if (useStorage) {
+        await storage.removeItem(item.file);
+      }
+    }
+
+    if (useStorage) {
       try {
-        await this.config.services.storage.setItem(file, JSON.stringify(event));
+        await storage.setItem(file, JSON.stringify(event));
       } catch (ex) {
-        this.config.services.log.error(`Error saving queue item to storage: ${ex.message}`)
+        log.error(`Error saving queue item to storage: ${ex.message}`)
       }
     }
 

packages/core/test/queue/DefaultEventQueue.test.ts

@@ -1,5 +1,6 @@
 import { Configuration } from "../../src/configuration/Configuration.js";
 import { Event } from "../../src/models/Event.js";
+import { DefaultEventQueue } from "../../src/queue/DefaultEventQueue.js";
 import { delay } from "../helpers.js";
 
 describe("DefaultEventQueue", () => {
@@ -60,4 +61,13 @@ describe("DefaultEventQueue", () => {
       expect(await config.services.storage.length()).toBe(1);
     }
   });
+
+  test("should respect max items", async () => {
+    config.services.queue = new DefaultEventQueue(config, 1);
+    const event: Event = { type: "log", reference_id: "123454321" };
+    for (let index = 0; index < 2; index++) {
+      await config.services.queue.enqueue(event);
+      expect(await config.services.storage.length()).toBe(1);
+    }
+  });
 });