diff --git a/packages/hydrooj/src/handler/judge.ts b/packages/hydrooj/src/handler/judge.ts index 8fb44b3b8..6a5d35858 100644 --- a/packages/hydrooj/src/handler/judge.ts +++ b/packages/hydrooj/src/handler/judge.ts @@ -18,7 +18,7 @@ import task from '../model/task'; import * as bus from '../service/bus'; import { updateJudge } from '../service/monitor'; import { - ConnectionHandler, Handler, post, Types, + ConnectionHandler, Handler, post, subscribe, Types, } from '../service/server'; import { sleep } from '../utils'; @@ -180,15 +180,14 @@ class JudgeConnectionHandler extends ConnectionHandler { async prepare() { logger.info('Judge daemon connected from ', this.request.ip); - this.send({ language: setting.langs }); - this.sendLanguageConfig = this.sendLanguageConfig.bind(this); - bus.on('system/setting', this.sendLanguageConfig); + this.sendLanguageConfig(); // Ensure language sent await sleep(100); this.newTask(); } - async sendLanguageConfig() { + @subscribe('system/setting') + sendLanguageConfig() { this.send({ language: setting.langs }); } @@ -226,7 +225,6 @@ class JudgeConnectionHandler extends ConnectionHandler { async cleanup() { logger.info('Judge daemon disconnected from ', this.request.ip); - bus.off('system/setting', this.sendLanguageConfig); if (this.processing) { await record.reset(this.processing.domainId, this.processing.rid, false); await task.add(this.processing); diff --git a/packages/hydrooj/src/handler/manage.ts b/packages/hydrooj/src/handler/manage.ts index 646bec10b..007a38983 100644 --- a/packages/hydrooj/src/handler/manage.ts +++ b/packages/hydrooj/src/handler/manage.ts @@ -14,7 +14,6 @@ import record from '../model/record'; import * as setting from '../model/setting'; import * as system from '../model/system'; import user from '../model/user'; -import * as bus from '../service/bus'; import { ConnectionHandler, Handler, param, requireSudo, Types, } from '../service/server'; @@ -176,7 +175,7 @@ class SystemSettingHandler extends SystemHandler { } } await Promise.all(tasks); - await bus.broadcast('system/setting', args); + this.ctx.broadcast('system/setting', args); this.back(); } } @@ -309,7 +308,7 @@ class SystemUserPrivHandler extends SystemHandler { const defaultPriv = system.get('default.priv'); await user.coll.updateMany({ priv: defaultPriv }, { $set: { priv } }); await system.set('default.priv', priv); - bus.broadcast('user/delcache', true); + this.ctx.broadcast('user/delcache', true); } this.back(); } diff --git a/packages/hydrooj/src/handler/record.ts b/packages/hydrooj/src/handler/record.ts index 7440bf0c8..1c83334a3 100644 --- a/packages/hydrooj/src/handler/record.ts +++ b/packages/hydrooj/src/handler/record.ts @@ -15,8 +15,9 @@ import storage from '../model/storage'; import * as system from '../model/system'; import TaskModel from '../model/task'; import user from '../model/user'; -import * as bus from '../service/bus'; -import { ConnectionHandler, param, Types } from '../service/server'; +import { + ConnectionHandler, param, subscribe, Types, +} from '../service/server'; import { buildProjection } from '../utils'; import { ContestDetailBaseHandler } from './contest'; import { postJudge } from './judge'; @@ -197,7 +198,7 @@ class RecordDetailHandler extends ContestDetailBaseHandler { const priority = await record.submissionPriority(this.user._id, -20); const isContest = this.rdoc.contest && this.rdoc.contest.toString() !== '000000000000000000000000'; const rdoc = await record.reset(domainId, rid, true); - bus.broadcast('record/change', rdoc); + this.ctx.broadcast('record/change', rdoc); await record.judge(domainId, rid, priority, isContest ? { detail: false } : {}); this.back(); } @@ -226,7 +227,6 @@ class RecordDetailHandler extends ContestDetailBaseHandler { } class RecordMainConnectionHandler extends ConnectionHandler { - cleanup: bus.Disposable = () => { }; all = false; tid: string; uid: number; @@ -273,7 +273,6 @@ class RecordMainConnectionHandler extends ConnectionHandler { this.checkPriv(PRIV.PRIV_MANAGE_ALL_DOMAIN); this.all = true; } - this.cleanup = bus.on('record/change', this.onRecordChange.bind(this)); } async message(msg: { rids: string[] }) { @@ -283,6 +282,7 @@ class RecordMainConnectionHandler extends ConnectionHandler { for (const rdoc of rdocs) this.onRecordChange(rdoc); } + @subscribe('record/change') async onRecordChange(rdoc: RecordDoc) { if (!this.all) { if (rdoc.domainId !== this.args.domainId) return; @@ -318,7 +318,6 @@ class RecordMainConnectionHandler extends ConnectionHandler { } class RecordDetailConnectionHandler extends ConnectionHandler { - cleanup: bus.Disposable = () => { }; pdoc: ProblemDoc; rid: string = ''; disconnectTimeout: NodeJS.Timeout; @@ -356,7 +355,6 @@ class RecordDetailConnectionHandler extends ConnectionHandler { this.pdoc = pdoc; this.throttleSend = throttle(this.sendUpdate, 1000, { trailing: true }); this.rid = rid.toString(); - this.cleanup = bus.on('record/change', this.onRecordChange.bind(this)); this.onRecordChange(rdoc); } @@ -367,6 +365,7 @@ class RecordDetailConnectionHandler extends ConnectionHandler { }); } + @subscribe('record/change') // eslint-disable-next-line async onRecordChange(rdoc: RecordDoc, $set?: any, $push?: any) { if (rdoc._id.toString() !== this.rid) return; diff --git a/packages/hydrooj/src/handler/training.ts b/packages/hydrooj/src/handler/training.ts index b3c589d7d..c6553c0af 100644 --- a/packages/hydrooj/src/handler/training.ts +++ b/packages/hydrooj/src/handler/training.ts @@ -15,7 +15,6 @@ import storage from '../model/storage'; import * as system from '../model/system'; import * as training from '../model/training'; import user from '../model/user'; -import * as bus from '../service/bus'; import { Handler, param, post, Types, } from '../service/server'; @@ -64,7 +63,7 @@ class TrainingMainHandler extends Handler { @param('page', Types.PositiveInt, true) async get(domainId: string, page = 1) { const query: FilterQuery = {}; - await bus.parallel('training/list', query, this); + await this.ctx.parallel('training/list', query, this); const [tdocs, tpcount] = await paginate( training.getMulti(domainId), page, @@ -102,7 +101,7 @@ class TrainingDetailHandler extends Handler { @param('uid', Types.PositiveInt, true) async get(domainId: string, tid: ObjectID, uid: number) { const tdoc = await training.get(domainId, tid); - await bus.parallel('training/get', tdoc, this); + await this.ctx.parallel('training/get', tdoc, this); let targetUser = this.user._id; let enrollUsers: number[] = []; let shouldCompare = false; diff --git a/packages/hydrooj/src/service/decorators.ts b/packages/hydrooj/src/service/decorators.ts index ff163e60b..ba8fb62db 100644 --- a/packages/hydrooj/src/service/decorators.ts +++ b/packages/hydrooj/src/service/decorators.ts @@ -4,7 +4,8 @@ import { Converter, Type, Validator } from '../lib/validator'; import { EventMap } from './bus'; import type { Handler } from './server'; -type MethodDecorator = (target: any, name: string, obj: any) => any; +type MethodDecorator = (target: any, funcName: string, obj: any) => any; +type ClassDecorator = any>(Class: T) => T extends new (...args: infer R) => infer S ? (...args: R) => S : never; export interface ParamOption { name: string, source: 'all' | 'get' | 'post' | 'route', @@ -82,10 +83,17 @@ export const post: DescriptorBuilder = (name, ...args) => _descriptor(_buildPara export const route: DescriptorBuilder = (name, ...args) => _descriptor(_buildParam(name, 'route', ...args)); export const param: DescriptorBuilder = (name, ...args) => _descriptor(_buildParam(name, 'all', ...args)); -export const subscribe: (name: keyof EventMap) => MethodDecorator = (name) => (target, funcName, obj) => { - target.__subscribe ||= []; - target.__subscribe.push({ name, target: obj.value }); - return obj; +export const subscribe: (name: keyof EventMap) => MethodDecorator | ClassDecorator = (name) => (target, funcName, obj) => { + if (funcName) { + target.__subscribe ||= []; + target.__subscribe.push({ name, target: obj.value }); + return obj; + } + return (...args) => { + const c = new target(...args); // eslint-disable-line new-cap + c.__subscribe = [{ name, target: c.send }]; + return c; + }; }; export function requireSudo(target: any, funcName: string, obj: any) {