f

package.json

@@ -24,6 +24,7 @@
     "cron"
   ],
   "dependencies": {
+    "@eggjs/utils": "^4.0.3",
     "cron-parser": "^4.9.0",
     "humanize-ms": "^2.0.0",
     "is-type-of": "^2.1.0",
@@ -38,6 +39,7 @@
     "@types/safe-timers": "^1.1.2",
     "egg": "beta",
     "egg-bin": "6",
+    "egg-logrotator": "^3.2.0",
     "egg-mock": "beta",
     "egg-tracer": "2",
     "eslint": "8",

src/agent.ts

@@ -25,11 +25,12 @@ export default class Boot implements ILifecycleBoot {
       // get job info from worker
       this.#agent.schedule.onJobFinish(info);
     });
+    debug('didLoad');
+  }
 
-    this.#agent.messenger.once('egg-ready', () => {
-      // start schedule after worker ready
-      this.#agent.schedule.start();
-      debug('got egg-ready event, schedule start');
-    });
+  async serverDidReady(): Promise<void> {
+    // start schedule after worker ready
+    this.#agent.schedule.start();
+    debug('serverDidReady, schedule start');
   }
 }

src/app.ts

@@ -3,6 +3,7 @@ import path from 'node:path';
 import type {
   Application, ILifecycleBoot, EggLogger,
 } from 'egg';
+import { importResolve } from '@eggjs/utils';
 import { ScheduleItem, ScheduleJobInfo } from './lib/types.js';
 
 const debug = debuglog('@eggjs/schedule/app');
@@ -16,15 +17,14 @@ export default class Boot implements ILifecycleBoot {
   }
 
   async didLoad(): Promise<void> {
-    debug('didLoad');
     const scheduleWorker = this.#app.scheduleWorker;
     await scheduleWorker.init();
 
     // log schedule list
     for (const s in scheduleWorker.scheduleItems) {
       const schedule = scheduleWorker.scheduleItems[s];
       if (!schedule.schedule.disable) {
-        this.#logger.info('[egg-schedule]: register schedule %s', schedule.key);
+        this.#logger.info('[@eggjs/schedule]: register schedule %s', schedule.key);
       }
     }
 
@@ -66,9 +66,9 @@ export default class Boot implements ILifecycleBoot {
           return await schedule.task(ctx, ...info.args);
         });
         success = true;
-      } catch (err) {
+      } catch (err: any) {
         success = false;
-        throw err;
+        e = err;
       }
 
       const rt = Date.now() - start;
@@ -97,28 +97,32 @@ export default class Boot implements ILifecycleBoot {
       ...config.schedule.directory,
     ];
     const runSchedule = async (schedulePath: string, ...args: any[]) => {
+      debug('[runSchedule] start schedulePath: %o, args: %o', schedulePath, args);
+
       // resolve real path
       if (path.isAbsolute(schedulePath)) {
-        schedulePath = require.resolve(schedulePath);
+        schedulePath = importResolve(schedulePath);
       } else {
         for (const dir of directory) {
+          const trySchedulePath = path.join(dir, schedulePath);
           try {
-            schedulePath = require.resolve(path.join(dir, schedulePath));
+            schedulePath = importResolve(trySchedulePath);
             break;
-          } catch (_) {
-            /* istanbul ignore next */
+          } catch (err) {
+            debug('[runSchedule] importResolve %o error: %s', trySchedulePath, err);
           }
         }
       }
 
+      debug('[runSchedule] resolve schedulePath: %o', schedulePath);
       let schedule: ScheduleItem;
       try {
         schedule = scheduleWorker.scheduleItems[schedulePath];
         if (!schedule) {
           throw new Error(`Cannot find schedule ${schedulePath}`);
         }
       } catch (err: any) {
-        err.message = `[egg-schedule] ${err.message}`;
+        err.message = `[@eggjs/schedule] ${err.message}`;
         throw err;
       }
 
@@ -132,5 +136,7 @@ export default class Boot implements ILifecycleBoot {
       });
     };
     Reflect.set(this.#app, 'runSchedule', runSchedule);
+
+    debug('didLoad');
   }
 }

src/lib/load_schedule.ts

@@ -1,7 +1,7 @@
 import path from 'node:path';
 import assert from 'node:assert';
 import { stringify } from 'node:querystring';
-import { isClass, isFunction } from 'is-type-of';
+import { isClass, isFunction, isGeneratorFunction } from 'is-type-of';
 import type { EggApplicationCore, EggContext } from 'egg';
 import type { ScheduleConfig, ScheduleTask, ScheduleItem } from './types.js';
 
@@ -20,20 +20,24 @@ function getScheduleLoader(app: EggApplicationCore) {
 
         let task: ScheduleTask;
         if (isClass(schedule)) {
+          assert(!isGeneratorFunction(schedule.prototype.subscribe),
+            `schedule(${fullpath}): "schedule" generator function is not support, should use async function instead`);
           task = async (ctx: EggContext, ...args: any[]) => {
             const instance = new schedule(ctx);
             // s.subscribe = app.toAsyncFunction(s.subscribe);
             return instance.subscribe(...args);
           };
         } else {
+          assert(!isGeneratorFunction(schedule.task),
+            `schedule(${fullpath}): "task" generator function is not support, should use async function instead`);
           task = schedule.task;
           // task = app.toAsyncFunction(schedule.task);
         }
 
         const env = app.config.env;
         const envList = schedule.schedule.env;
         if (Array.isArray(envList) && !envList.includes(env)) {
-          app.coreLogger.info(`[egg-schedule]: ignore schedule ${fullpath} due to \`schedule.env\` not match`);
+          app.coreLogger.info(`[@eggjs/schedule]: ignore schedule ${fullpath} due to \`schedule.env\` not match`);
           continue;
         }
 

src/lib/schedule.ts

@@ -1,8 +1,11 @@
+import { debuglog } from 'node:util';
 import type { Agent, EggLogger } from 'egg';
 import { loadSchedule } from './load_schedule.js';
 import type { ScheduleItem, ScheduleJobInfo } from './types.js';
 import type { BaseStrategy } from './strategy/base.js';
 
+const debug = debuglog('@eggjs/schedule/lib/schedule');
+
 export class Schedule {
   closed = false;
 
@@ -23,6 +26,7 @@ export class Schedule {
    */
   use(type: string, clz: typeof BaseStrategy) {
     this.#strategyClassMap.set(type, clz);
+    debug('use type: %o', type);
   }
 
   /**
@@ -38,7 +42,9 @@ export class Schedule {
   registerSchedule(scheduleItem: ScheduleItem) {
     const { key, schedule } = scheduleItem;
     const type = schedule.type;
-    if (schedule.disable) return;
+    if (schedule.disable) {
+      return;
+    }
 
     // find Strategy by type
     const Strategy = this.#strategyClassMap.get(type!);
@@ -51,9 +57,11 @@ export class Schedule {
     // Initialize strategy and register
     const instance = new Strategy(schedule, this.#agent, key);
     this.#strategyInstanceMap.set(key, instance);
+    debug('registerSchedule type: %o, config: %o, key: %o', type, schedule, key);
   }
 
   unregisterSchedule(key: string) {
+    debug('unregisterSchedule key: %o', key);
     return this.#strategyInstanceMap.delete(key);
   }
 
@@ -75,6 +83,7 @@ export class Schedule {
    * start schedule
    */
   start() {
+    debug('start');
     this.closed = false;
     for (const instance of this.#strategyInstanceMap.values()) {
       instance.start();
@@ -86,5 +95,6 @@ export class Schedule {
     for (const instance of this.#strategyInstanceMap.values()) {
       instance.close();
     }
+    debug('close');
   }
 }

src/lib/schedule_worker.ts

@@ -4,7 +4,7 @@ import type { ScheduleItem } from './types.js';
 
 export class ScheduleWorker {
   #app: Application;
-  scheduleItems: Record<string, ScheduleItem>;
+  scheduleItems: Record<string, ScheduleItem> = {};
 
   constructor(app: Application) {
     this.#app = app;

src/lib/strategy/timer.ts

@@ -15,22 +15,22 @@ export abstract class TimerStrategy extends BaseStrategy {
 
     const { interval, cron, cronOptions, immediate } = this.scheduleConfig;
     assert(interval || cron || immediate,
-      `[egg-schedule] ${this.key} \`schedule.interval\` or \`schedule.cron]\` or \`schedule.immediate\` must be present`);
+      `[@eggjs/schedule] ${this.key} \`schedule.interval\` or \`schedule.cron\` or \`schedule.immediate\` must be present`);
 
     // init cron parser
     if (cron) {
       try {
         this.cronInstance = parseExpression(cron, cronOptions);
       } catch (err: any) {
         throw new TypeError(
-          `[egg-schedule] ${this.key} parse cron instruction(${cron}) error: ${err.message}`,
+          `[@eggjs/schedule] ${this.key} parse cron instruction(${cron}) error: ${err.message}`,
           { cause: err });
       }
     }
   }
 
   protected handler() {
-    throw new TypeError(`[egg-schedule] ${this.key} strategy should override \`handler()\` method`);
+    throw new TypeError(`[@eggjs/schedule] ${this.key} strategy should override \`handler()\` method`);
   }
 
 

test/fixtures/customTypeParams/app/schedule/cluster-all-clz.js

@@ -10,7 +10,7 @@ class Interval extends Subscription {
     };
   }
 
-  * subscribe(data) {
+  async subscribe(data) {
     this.ctx.logger.info('cluster_all_log_clz', data);
   }
 }

test/fixtures/customTypeParams/app/schedule/cluster-clz.js

@@ -10,7 +10,7 @@ class Interval extends Subscription {
     };
   }
 
-  * subscribe(data) {
+  async subscribe(data) {
     this.ctx.logger.info('cluster_log_clz', data);
   }
 }

test/fixtures/executeError-task-generator/app/schedule/interval.js

@@ -0,0 +1,10 @@
+'use strict';
+
+exports.schedule = {
+  type: 'worker',
+  interval: 2000,
+};
+
+exports.task = function* () {
+  throw new Error('interval error');
+};

test/fixtures/executeError-task-generator/package.json

@@ -0,0 +1,3 @@
+{
+  "name": "executeError"
+}

test/fixtures/executeError/app/schedule/interval.js

@@ -1,10 +1,8 @@
-'use strict';
-
 exports.schedule = {
   type: 'worker',
   interval: 2000,
 };
 
-exports.task = function* () {
+exports.task = async function() {
   throw new Error('interval error');
 };

test/fixtures/scheduleError/app/schedule/interval.js

@@ -1,5 +1,3 @@
-'use strict';
-
 exports.schedule = {
   type: 'worker',
 };

test/fixtures/scheduleError/app/schedule/sub/cron.js

@@ -1,5 +1,3 @@
-'use strict';
-
 exports.schedule = {
   type: 'worker',
   cron: '*/5 * * * * *',

test/fixtures/subscription-enableFastContextLogger/app/schedule/interval.js

@@ -8,7 +8,7 @@ class Interval extends Subscription {
     };
   }
 
-  * subscribe() {
+  async subscribe() {
     this.app.logger.info('interval');
   }
 }

test/fixtures/subscription-enableFastContextLogger/app/schedule/sub/cron.js

@@ -8,7 +8,7 @@ class Interval extends Subscription {
     };
   }
 
-  * subscribe() {
+  async subscribe() {
     this.app.logger.info('cron');
   }
 }

test/fixtures/subscription-generator/app/schedule/interval.js

@@ -0,0 +1,16 @@
+const Subscription = require('egg').Subscription;
+
+class Interval extends Subscription {
+  static get schedule() {
+    return {
+      type: 'worker',
+      interval: 4000,
+    };
+  }
+
+  * subscribe() {
+    this.ctx.logger.info('interval');
+  }
+}
+
+module.exports = Interval;

test/fixtures/subscription-generator/app/schedule/sub/cron.js

@@ -0,0 +1,16 @@
+const Subscription = require('egg').Subscription;
+
+class Interval extends Subscription {
+  static get schedule() {
+    return {
+      type: 'worker',
+      cron: '*/5 * * * * *',
+    };
+  }
+
+  async subscribe() {
+    this.ctx.logger.info('cron');
+  }
+}
+
+module.exports = Interval;

test/fixtures/subscription-generator/config/plugin.js

@@ -0,0 +1 @@
+exports.logrotator = true;

test/fixtures/subscription-generator/package.json

@@ -0,0 +1,3 @@
+{
+  "name": "subscription"
+}