add queue retry

src/queue-runner.ts

@@ -2,11 +2,13 @@ import { Aragami, ClassType } from 'aragami';
 import { ConsoleLogger, Logger } from '@nestjs/common';
 import * as os from 'os';
 import { InjectAragami } from '../index';
+import { retry } from 'rxjs';
 
 export interface QueueRunnerOptions {
   maxConcurrency?: number;
   key?: string;
   logTask?: boolean;
+  retry?: number;
 }
 
 export class _QueueRunner<T> {
@@ -43,6 +45,8 @@ export class _QueueRunner<T> {
     return;
   }
 
+  _queueKey = this._queueRunnerOptions.key || 'default';
+
   async _mainLoop(i: number) {
     const logger = new ConsoleLogger(`${this._queueClass.name}-${i}`);
     while (!this._quit) {
@@ -50,13 +54,31 @@ export class _QueueRunner<T> {
       try {
         const task: T = await this.aragami.queueGatherBlocking(
           this._queueClass,
-          this._queueRunnerOptions.key || 'default',
+          this._queueKey,
         );
         if (!task) continue;
         if (this._queueRunnerOptions.logTask) {
           logger.log(`Got task: ${JSON.stringify(task)}`);
         }
-        await this.runTask(task, i);
+        try {
+          await this.runTask(task, i);
+        } catch (e) {
+          logger.error(`Task failed: ${e}`);
+          if (
+            this._queueRunnerOptions.retry &&
+            (!task['__retry'] ||
+              task['__retry'] < this._queueRunnerOptions.retry)
+          ) {
+            logger.log(`Retrying task: ${JSON.stringify(task)}`);
+            if (this._queueRunnerOptions.retry > 0) {
+              task['__retry'] = (task['__retry'] || 0) + 1;
+            }
+            await this.aragami.queueAdd(task, {
+              key: this._queueKey,
+              prior: true,
+            });
+          }
+        }
       } catch (e) {
         logger.error(`Loop failed: ${e}`);
       }