diff --git a/.eslintrc.js b/.eslintrc.js index 394ebf64a..c3e5e957c 100644 --- a/.eslintrc.js +++ b/.eslintrc.js @@ -29,6 +29,7 @@ module.exports = { 'max-len': ['warn', 150], 'no-bitwise': 'off', 'no-console': 'off', + 'no-continue': 'off', 'no-dupe-class-members': 'off', 'no-extend-native': 'off', 'no-inner-declarations': 'off', diff --git a/packages/hydrooj/TODO b/packages/hydrooj/TODO index a8c8931e2..3ef3e74d9 100644 --- a/packages/hydrooj/TODO +++ b/packages/hydrooj/TODO @@ -4,4 +4,5 @@ SETTINGS_PRIVACY.allowRegisteredUsers model.setting.ui.name pdoc.secretConfig? message.markAsRead -基于document的model难以lint,建议拆分 \ No newline at end of file +基于document的model难以lint,建议拆分 +检查多余的索引 \ No newline at end of file diff --git a/packages/hydrooj/package.json b/packages/hydrooj/package.json index 54fcea715..95e6ed9fc 100644 --- a/packages/hydrooj/package.json +++ b/packages/hydrooj/package.json @@ -1,6 +1,6 @@ { "name": "hydrooj", - "version": "2.13.0", + "version": "2.13.1", "bin": "bin/hydrooj.js", "main": "dist/loader.js", "typings": "dist/loader.d.ts", diff --git a/packages/hydrooj/src/entry/cli.ts b/packages/hydrooj/src/entry/cli.ts index 0534c8185..275b820ce 100644 --- a/packages/hydrooj/src/entry/cli.ts +++ b/packages/hydrooj/src/entry/cli.ts @@ -5,6 +5,7 @@ import { lib, service, model, builtinLib, builtinModel, } from './common'; +import * as bus from '../service/bus'; const COMMENTS = /((\/\/.*$)|(\/\*[\s\S]*?\*\/))/mg; const ARR = /=>.*$/mg; @@ -51,13 +52,8 @@ export async function load() { require('../utils'); require('../error'); require('../options'); - const bus = require('../service/bus'); await new Promise((resolve) => { - const h = () => { - bus.unsubscribe(['system_database_connected'], h); - resolve(); - }; - bus.subscribe(['system_database_connected'], h); + bus.once('database/connect', resolve); require('../service/db'); }); for (const i of builtinLib) require(`../lib/${i}`); @@ -66,6 +62,6 @@ export async function load() { await service(pending, fail); for (const i of builtinModel) require(`../model/${i}`); await model(pending, fail); - for (const postInit of global.Hydro.postInit) await postInit(); + await bus.parallel('app/started'); await cli(); } diff --git a/packages/hydrooj/src/entry/master.ts b/packages/hydrooj/src/entry/master.ts index 3a73af1ca..25060e6a6 100644 --- a/packages/hydrooj/src/entry/master.ts +++ b/packages/hydrooj/src/entry/master.ts @@ -7,6 +7,7 @@ import pslist from 'ps-list'; import { argv } from 'yargs'; import { builtinModel } from './common'; import { Entry } from '../loader'; +import * as bus from '../service/bus'; const tmpdir = path.resolve(os.tmpdir(), 'hydro'); const lockfile = path.resolve(tmpdir, 'lock.json'); @@ -31,9 +32,6 @@ export async function load(call: Entry) { ppid: process.ppid, }; await fs.writeFile(lockfile, JSON.stringify(context)); - global.onDestory.push(() => { - fs.removeSync(lockfile); - }); require('../lib/i18n'); require('../utils'); require('../error'); @@ -45,14 +43,15 @@ export async function load(call: Entry) { process.exit(1); }); } - const bus = require('../service/bus'); + bus.once('app/exit', () => { + fs.removeSync(lockfile); + }); await new Promise((resolve) => { const h = () => { console.log('Database connected'); - bus.unsubscribe(['system_database_connected'], h); resolve(); }; - bus.subscribe(['system_database_connected'], h); + bus.once('database/connect', h); require('../service/db'); }); require('../service/monitor'); @@ -63,11 +62,6 @@ export async function load(call: Entry) { const ins = require('../script/upgrade0_1'); await ins.run({ username: 'Root', password: 'rootroot' }); } - for (const postInit of global.Hydro.postInit) { - await postInit().catch((e) => { - console.error('Error running postInit:', postInit.toString()); - console.error(e); - }); - } + bus.serial('app/started'); return await modelSystem.get('server.worker'); } diff --git a/packages/hydrooj/src/entry/worker.ts b/packages/hydrooj/src/entry/worker.ts index d10886382..c1d757409 100644 --- a/packages/hydrooj/src/entry/worker.ts +++ b/packages/hydrooj/src/entry/worker.ts @@ -4,6 +4,7 @@ import { locale, template, lib, service, model, handler, script, setting, uistatic, builtinLib, builtinScript, builtinHandler, builtinModel, } from './common'; +import * as bus from '../service/bus'; export async function load() { const pending = global.addons; @@ -18,14 +19,12 @@ export async function load() { template(pending, fail), uistatic(pending, fail), ]); - const bus = require('../service/bus'); await new Promise((resolve) => { const h = () => { console.log('Database connected'); - bus.unsubscribe(['system_database_connected'], h); resolve(); }; - bus.subscribe(['system_database_connected'], h); + bus.once('database/connect', h); require('../service/db'); }); for (const i of builtinLib) require(`../lib/${i}`); @@ -48,7 +47,7 @@ export async function load() { await notfound.apply(); for (const i of builtinScript) require(`../script/${i}`); await script(pending, fail, active); - for (const postInit of global.Hydro.postInit) await postInit(); + await bus.serial('app/started'); await server.start(); setInterval(() => { process.send({ event: 'stat', count: global.Hydro.stat.reqCount }); diff --git a/packages/hydrooj/src/handler/contest.ts b/packages/hydrooj/src/handler/contest.ts index 5882c1415..09eb6f884 100644 --- a/packages/hydrooj/src/handler/contest.ts +++ b/packages/hydrooj/src/handler/contest.ts @@ -310,7 +310,7 @@ class ContestProblemSubmitHandler extends ContestProblemHandler { record.get(domainId, rid), contest.updateStatus(domainId, this.tdoc.docId, this.user._id, rid, this.pdoc.docId), ]); - bus.publish('record_change', rdoc); + bus.boardcast('record/change', rdoc); if (!this.canShowRecord(this.tdoc)) { this.response.body = { tid }; this.response.redirect = this.url('contest_detail', { tid }); diff --git a/packages/hydrooj/src/handler/home.ts b/packages/hydrooj/src/handler/home.ts index 7712c1105..17436f5ad 100644 --- a/packages/hydrooj/src/handler/home.ts +++ b/packages/hydrooj/src/handler/home.ts @@ -25,6 +25,7 @@ import { PERM, PRIV, CONSTANT } from '../model/builtin'; import { isContent, isPassword, isEmail, isTitle, } from '../lib/validator'; +import { Mdoc } from '../interface'; const { geoip, useragent } = global.Hydro.lib; @@ -358,18 +359,19 @@ class HomeMessagesHandler extends Handler { } class HomeMessagesConnectionHandler extends ConnectionHandler { - id: string; + dispose: bus.Disposable; async prepare() { - bus.subscribe([`user_message-${this.user._id}`], this.onMessageReceived.bind(this)); + this.dispose = bus.on('user/message', this.onMessageReceived.bind(this)); } - async onMessageReceived(e: any) { - this.send(e.value); + async onMessageReceived(uid: number, e: any) { + if (uid !== this.user._id) return; + this.send(e); } async cleanup() { - bus.unsubscribe([`user_message-${this.user._id}`], this.id); + if (this.dispose) this.dispose(); } } diff --git a/packages/hydrooj/src/handler/homework.ts b/packages/hydrooj/src/handler/homework.ts index b02bab638..2d282f2b7 100644 --- a/packages/hydrooj/src/handler/homework.ts +++ b/packages/hydrooj/src/handler/homework.ts @@ -189,7 +189,7 @@ class HomeworkDetailProblemSubmitHandler extends HomeworkDetailProblemHandler { contest.updateStatus(domainId, tid, this.user._id, rid, pid, false, 0, document.TYPE_HOMEWORK), ]); - bus.publish('record_change', rdoc); + bus.boardcast('record/change', rdoc); this.response.body = { tid, rid }; if (this.canShowRecord(this.tdoc)) this.response.redirect = this.url('record_detail', { rid }); else this.response.redirect = this.url('homework_detail', { tid }); diff --git a/packages/hydrooj/src/handler/judge.ts b/packages/hydrooj/src/handler/judge.ts index ecd791a8f..be2a5d19d 100644 --- a/packages/hydrooj/src/handler/judge.ts +++ b/packages/hydrooj/src/handler/judge.ts @@ -62,7 +62,7 @@ export async function next(body: JudgeResultBody) { if (body.memory) $set.memory = body.memory; if (body.progress) $set.progress = body.progress; rdoc = await record.update(body.domainId, body.rid, $set, $push); - bus.publish('record_change', { rdoc, $set, $push }); + bus.boardcast('record/change', rdoc, $set, $push); } export async function end(body: JudgeResultBody) { @@ -87,7 +87,7 @@ export async function end(body: JudgeResultBody) { $set.judger = body.judger; rdoc = await record.update(body.domainId, body.rid, $set, $push, $unset); await _postJudge(rdoc); - bus.publish('record_change', { rdoc }); // trigger a full update + bus.boardcast('record/change', rdoc); // trigger a full update } class JudgeHandler extends Handler { diff --git a/packages/hydrooj/src/handler/problem.ts b/packages/hydrooj/src/handler/problem.ts index 4e91669bf..f711318d0 100644 --- a/packages/hydrooj/src/handler/problem.ts +++ b/packages/hydrooj/src/handler/problem.ts @@ -1,6 +1,6 @@ import { isSafeInteger, flatten } from 'lodash'; import yaml from 'js-yaml'; -import { ObjectID } from 'mongodb'; +import { FilterQuery, ObjectID } from 'mongodb'; import { NoProblemError, ProblemDataNotFoundError, BadRequestError, SolutionNotFoundError, ProblemNotFoundError, ValidationError, @@ -61,16 +61,14 @@ class ProblemMainHandler extends ProblemHandler { @param('q', Types.String, true) async get(domainId: string, page = 1, q = '') { this.response.template = 'problem_main.html'; - const query: any = {}; + const query: FilterQuery = {}; let psdict = {}; const path: PathComponent[] = [ ['Hydro', 'homepage'], ['problem_main', null], ]; if (q) { - q = q.toLowerCase(); - const $regex = new RegExp(`\\A\\Q${q}\\E`, 'gmi'); - query.title = { $regex }; + query.$text = { $search: q }; path.push([q, null, null, true]); } if (!this.user.hasPerm(PERM.PERM_VIEW_PROBLEM_HIDDEN)) query.hidden = false; @@ -252,7 +250,7 @@ class ProblemSubmitHandler extends ProblemDetailHandler { record.get(domainId, rid), domain.incUserInDomain(domainId, this.user._id, 'nSubmit'), ]); - bus.publish('record_change', { rdoc }); + bus.boardcast('record/change', rdoc); this.response.body = { rid }; this.response.redirect = this.url('record_detail', { rid }); } @@ -275,7 +273,7 @@ class ProblemPretestHandler extends ProblemDetailHandler { }, ); const rdoc = await record.get(domainId, rid); - bus.publish('record_change', { rdoc }); + bus.boardcast('record/change', rdoc); this.response.body = { rid }; } } @@ -285,7 +283,7 @@ class ProblemPretestConnectionHandler extends ConnectionHandler { domainId: string; - id: string; + dispose: bus.Disposable; @param('pid', Types.String) async prepare(domainId: string, pid: string) { @@ -294,11 +292,10 @@ class ProblemPretestConnectionHandler extends ConnectionHandler { if (!pdoc) throw new ProblemNotFoundError(domainId, pid); this.pid = pdoc.docId.toString(); this.domainId = domainId; - this.id = bus.subscribe(['record_change'], this.onRecordChange.bind(this)); + this.dispose = bus.on('record/change', this.onRecordChange.bind(this)); } - async onRecordChange(data) { - const rdoc: Rdoc = data.value.rdoc; + async onRecordChange(rdoc: Rdoc) { if ( rdoc.uid !== this.user._id || rdoc.pid.toString() !== this.pid @@ -316,7 +313,7 @@ class ProblemPretestConnectionHandler extends ConnectionHandler { } async cleanup() { - bus.unsubscribe(['record_change'], this.id); + if (this.dispose) this.dispose(); } } diff --git a/packages/hydrooj/src/handler/record.ts b/packages/hydrooj/src/handler/record.ts index 57d6074c3..f0e65087f 100644 --- a/packages/hydrooj/src/handler/record.ts +++ b/packages/hydrooj/src/handler/record.ts @@ -102,6 +102,8 @@ class RecordRejudgeHandler extends Handler { const RecordConnectionHandler = contest.ContestHandlerMixin(ConnectionHandler); class RecordMainConnectionHandler extends RecordConnectionHandler { + dispose: bus.Disposable; + @param('tid', Types.ObjectID, true) async prepare(domainId: string, tid: ObjectID) { this.domainId = domainId; @@ -113,7 +115,7 @@ class RecordMainConnectionHandler extends RecordConnectionHandler { return; } } - this.id = bus.subscribe(['record_change'], this.onRecordChange.bind(this)); + this.dispose = bus.on('record/change', this.onRecordChange.bind(this)); } async message(msg) { @@ -123,17 +125,16 @@ class RecordMainConnectionHandler extends RecordConnectionHandler { this.domainId, { _id: { $in: rids } }, ).toArray(); for (const rdoc of rdocs) { - this.onRecordChange({ value: { rdoc } }); + this.onRecordChange(rdoc); } } } async cleanup() { - bus.unsubscribe(['record_change'], this.id); + if (this.dispose) this.dispose(); } - async onRecordChange(data) { - const rdoc: Rdoc = data.value.rdoc; + async onRecordChange(rdoc: Rdoc) { if (rdoc.contest && rdoc.contest.tid.toString() !== this.tid) return; // eslint-disable-next-line prefer-const let [udoc, pdoc] = await Promise.all([ @@ -146,6 +147,8 @@ class RecordMainConnectionHandler extends RecordConnectionHandler { } class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(ConnectionHandler) { + dispose: bus.Disposable; + @param('rid', Types.ObjectID) async prepare(domainId: string, rid: ObjectID) { const rdoc = await record.get(domainId, rid); @@ -157,13 +160,11 @@ class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(Connecti } } this.rid = rid.toString(); - this.id = bus.subscribe(['record_change'], this.onRecordChange.bind(this)); - this.onRecordChange({ value: { rdoc } }); + this.dispose = bus.on('record/change', this.onRecordChange.bind(this)); + this.onRecordChange(rdoc); } - async onRecordChange(data) { - const rdoc: Rdoc = data.value.rdoc; - const { $set, $push } = data.value; + async onRecordChange(rdoc: Rdoc, $set?: any, $push?: any) { if (rdoc._id.toString() !== this.rid) return; if ($set) this.send({ $set, $push }); else { @@ -175,7 +176,7 @@ class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(Connecti } async cleanup() { - bus.unsubscribe(['record_change'], this.id); + if (this.dispose) this.dispose(); } } diff --git a/packages/hydrooj/src/interface.ts b/packages/hydrooj/src/interface.ts index fb610c5a9..64af21769 100644 --- a/packages/hydrooj/src/interface.ts +++ b/packages/hydrooj/src/interface.ts @@ -455,9 +455,7 @@ declare global { ui: UI, error: typeof import('./error'), locales: Dict>, - postInit: Array<() => Promise>, }, - onDestory: Array<() => void | Promise>, addons: string[], } } diff --git a/packages/hydrooj/src/loader.ts b/packages/hydrooj/src/loader.ts index 3b4f20571..f959bd597 100644 --- a/packages/hydrooj/src/loader.ts +++ b/packages/hydrooj/src/loader.ts @@ -36,9 +36,7 @@ if (!global.Hydro) { // @ts-ignore error: {}, locales: {}, - postInit: [], }; - global.onDestory = []; global.addons = []; } @@ -47,9 +45,12 @@ if (argv.debug) { console.log(argv); } +// eslint-disable-next-line import/first +import * as bus from './service/bus'; + async function terminate() { try { - for (const task of global.onDestory) await task(); + await bus.parallel('app/exit'); } catch (e) { process.exit(1); } @@ -127,7 +128,7 @@ async function messageHandler(worker: cluster.Worker, msg: any) { cluster.workers[i].send(msg); } } else { - global.Hydro.service.bus.publish(msg.eventName, msg.payload, false); + global.Hydro.service.bus.emit(msg.eventName, ...msg.payload); } } else if (msg.event === 'stat') { global.Hydro.stat.reqCount += msg.count; diff --git a/packages/hydrooj/src/model/blacklist.ts b/packages/hydrooj/src/model/blacklist.ts index e5366d9fb..54e69a49e 100644 --- a/packages/hydrooj/src/model/blacklist.ts +++ b/packages/hydrooj/src/model/blacklist.ts @@ -1,5 +1,6 @@ import { Collection } from 'mongodb'; import * as db from '../service/db'; +import * as bus from '../service/bus'; import { Bdoc } from '../interface'; const coll: Collection = db.collection('blacklist'); @@ -36,5 +37,5 @@ function ensureIndexes() { return coll.createIndex('expireAt', { expireAfterSeconds: 0 }); } -global.Hydro.postInit.push(ensureIndexes); +bus.once('app/started', ensureIndexes); global.Hydro.model.blacklist = { add, get, del }; diff --git a/packages/hydrooj/src/model/document.ts b/packages/hydrooj/src/model/document.ts index 3dd5502fa..9dd70141c 100644 --- a/packages/hydrooj/src/model/document.ts +++ b/packages/hydrooj/src/model/document.ts @@ -1,11 +1,12 @@ import assert from 'assert'; import { ObjectID } from 'mongodb'; import * as db from '../service/db'; +import * as bus from '../service/bus'; type DocID = ObjectID | string | number; -const coll = db.collection('document'); -const collStatus = db.collection('document.status'); +export const coll = db.collection('document'); +export const collStatus = db.collection('document.status'); export const TYPE_PROBLEM = 10; export const TYPE_PROBLEM_SOLUTION = 11; @@ -51,6 +52,7 @@ export async function add( doc.parentType = parentType; doc.parentId = parentId; } + await bus.serial('document/add', doc); const res = await coll.insertOne(doc); return docId || res.insertedId; } @@ -337,6 +339,7 @@ async function ensureIndexes() { await ic({ domainId: 1, docType: 1, docId: 1 }, u); await ic({ domainId: 1, docType: 1, owner: 1, docId: -1 }); // For problem + await ic({ domainId: 1, docType: 1, title: 'text' }, s); await ic({ domainId: 1, docType: 1, category: 1, docId: 1 }, s); await ic({ domainId: 1, docType: 1, hidden: 1, category: 1, docId: 1 }, s); await ic({ domainId: 1, docType: 1, tag: 1, docId: 1 }, s); @@ -364,8 +367,11 @@ async function ensureIndexes() { await is({ domainId: 1, docType: 1, uid: 1, enroll: 1, docId: 1 }, s); } -global.Hydro.postInit.push(ensureIndexes); +bus.once('app/started', ensureIndexes); global.Hydro.model.document = { + coll, + collStatus, + add, addToSet, cappedIncStatus, diff --git a/packages/hydrooj/src/model/file.ts b/packages/hydrooj/src/model/file.ts index 8f000974a..f5c05e28a 100644 --- a/packages/hydrooj/src/model/file.ts +++ b/packages/hydrooj/src/model/file.ts @@ -2,10 +2,11 @@ import { ObjectID, Collection } from 'mongodb'; import * as fs from 'fs'; import { Dictionary } from 'lodash'; import { ForbiddenError, NotFoundError } from '../error'; +import { Ufdoc } from '../interface'; import * as db from '../service/db'; +import * as bus from '../service/bus'; import gridfs from '../service/gridfs'; import hash from '../lib/hash.hydro'; -import { Ufdoc } from '../interface'; const coll: Collection = db.collection('file'); const collFile = db.collection('fs.files'); @@ -121,7 +122,7 @@ function ensureIndexes() { ]); } -global.Hydro.postInit.push(ensureIndexes); +bus.once('app/started', ensureIndexes); global.Hydro.model.file = { add, get, del, getMeta, getWithSecret, getMulti, getMetaDict, url, }; diff --git a/packages/hydrooj/src/model/message.ts b/packages/hydrooj/src/model/message.ts index 84a7b7988..a21d907e8 100644 --- a/packages/hydrooj/src/model/message.ts +++ b/packages/hydrooj/src/model/message.ts @@ -22,7 +22,7 @@ export async function send( if (from !== to) { // ENHANCE domainId? const udoc = await user.getById('system', to); - bus.publish(`user_message-${to}`, { mdoc, udoc }); + bus.boardcast('user/message', to, mdoc, udoc); } return mdoc; } @@ -69,7 +69,7 @@ function ensureIndexes() { ]); } -global.Hydro.postInit.push(ensureIndexes); +bus.once('app/started', ensureIndexes); global.Hydro.model.message = { FLAG_UNREAD, FLAG_ALERT, diff --git a/packages/hydrooj/src/model/opcount.ts b/packages/hydrooj/src/model/opcount.ts index f9bd53ca7..0acc17e11 100644 --- a/packages/hydrooj/src/model/opcount.ts +++ b/packages/hydrooj/src/model/opcount.ts @@ -1,4 +1,5 @@ import { OpcountExceededError } from '../error'; +import * as bus from '../service/bus'; import * as db from '../service/db'; const coll = db.collection('opcount'); @@ -23,5 +24,5 @@ function ensureIndexes() { return coll.createIndex('expireAt', { expireAfterSeconds: 0 }); } -global.Hydro.postInit.push(ensureIndexes); +bus.once('app/started', ensureIndexes); global.Hydro.model.opcount = { inc }; diff --git a/packages/hydrooj/src/model/setting.ts b/packages/hydrooj/src/model/setting.ts index 6cf6b79ac..a3797a43a 100644 --- a/packages/hydrooj/src/model/setting.ts +++ b/packages/hydrooj/src/model/setting.ts @@ -3,6 +3,7 @@ import moment from 'moment-timezone'; import { Dictionary } from 'lodash'; import * as builtin from './builtin'; import { Setting as _Setting } from '../interface'; +import * as bus from '../service/bus'; type SettingDict = Dictionary<_Setting>; @@ -139,18 +140,16 @@ SystemSetting( Setting('setting_storage', 'user', null, 1, 'number', 'User Counter', null, FLAG_DISABLED | FLAG_HIDDEN), ); -global.Hydro.postInit.push( - async () => { - for (const setting of SYSTEM_SETTINGS) { - if (setting.value) { - const current = await global.Hydro.model.system.get(setting.key); - if (current === null || current === '') { - await global.Hydro.model.system.set(setting.key, setting.value); - } +bus.once('app/started', async () => { + for (const setting of SYSTEM_SETTINGS) { + if (setting.value) { + const current = await global.Hydro.model.system.get(setting.key); + if (current === null || current === '') { + await global.Hydro.model.system.set(setting.key, setting.value); } } - }, -); + } +}); global.Hydro.model.setting = { langRange, diff --git a/packages/hydrooj/src/model/token.ts b/packages/hydrooj/src/model/token.ts index b2cfe2060..f508ce5c6 100644 --- a/packages/hydrooj/src/model/token.ts +++ b/packages/hydrooj/src/model/token.ts @@ -1,6 +1,7 @@ import { Collection } from 'mongodb'; import { TokenDoc } from '../interface'; import * as db from '../service/db'; +import * as bus from '../service/bus'; export const coll: Collection = db.collection('token'); @@ -87,7 +88,7 @@ export function delByUid(uid: number) { return coll.deleteMany({ uid }); } -global.Hydro.postInit.push(ensureIndexes); +bus.once('app/started', ensureIndexes); global.Hydro.model.token = { coll, diff --git a/packages/hydrooj/src/model/user.ts b/packages/hydrooj/src/model/user.ts index c029dc263..4d98621b9 100644 --- a/packages/hydrooj/src/model/user.ts +++ b/packages/hydrooj/src/model/user.ts @@ -5,9 +5,10 @@ import * as setting from './setting'; import * as domain from './domain'; import { BUILTIN_USERS, PRIV } from './builtin'; import { UserNotFoundError, UserAlreadyExistError, LoginError } from '../error'; +import { User as _User, Udoc, Udict } from '../interface'; import pwhash from '../lib/hash.hydro'; import * as db from '../service/db'; -import { User as _User, Udoc, Udict } from '../interface'; +import * as bus from '../service/bus'; const coll: Collection = db.collection('user'); @@ -248,7 +249,7 @@ function ensureIndexes() { ]); } -global.Hydro.postInit.push(ensureIndexes); +bus.once('app/started', ensureIndexes); global.Hydro.model.user = { User, create, diff --git a/packages/hydrooj/src/service/bus.ts b/packages/hydrooj/src/service/bus.ts index 1c2553631..587b41de5 100644 --- a/packages/hydrooj/src/service/bus.ts +++ b/packages/hydrooj/src/service/bus.ts @@ -1,41 +1,122 @@ +/* eslint-disable no-await-in-loop */ import cluster from 'cluster'; +import { Db } from 'mongodb'; +import { Mdoc, Rdoc, User } from '../interface'; -const bus = {}; +const _hooks: Record any>> = {}; +const _disposables = []; -export function subscribe(events: string[], handler: any) { - const id = String.random(16); - for (const event of events) { - if (!bus[event]) bus[event] = {}; - bus[event][id] = handler; - } - return id; +function isBailed(value: any) { + return value !== null && value !== false && value !== undefined; } -export function unsubscribe(events: string[], id: string) { - for (const event of events) { - if (!bus[event]) bus[event] = {}; - delete bus[event][id]; - } +export type Disposable = () => void + +interface EventMap { + 'app/started': () => void + 'app/exit': () => Promise | void + 'dispose': () => void + + 'database/connect': (db: Db) => void + + 'user/message': (uid: number, mdoc: Mdoc, udoc: User) => void + + 'document/add': (doc: any) => string | void + + 'record/change': (rdoc: Rdoc, $set?: any, $push?: any) => void } -export function publish(event: string, payload: any, isMaster = true) { +function getHooks(name: K) { + const hooks = _hooks[name] || (_hooks[name] = []); + if (hooks.length >= 128) { + console.warn( + 'max listener count (128) for event "%s" exceeded, which may be caused by a memory leak', + name, + ); + } + return hooks; +} + +export function removeListener(name: K, listener: EventMap[K]) { + const index = (_hooks[name] || []).findIndex((callback) => callback === listener); + if (index >= 0) { + _hooks[name].splice(index, 1); + return true; + } + return false; +} + +export function addListener(name: K, listener: EventMap[K]) { + getHooks(name).push(listener); + const dispose = () => removeListener(name, listener); + _disposables.push(name === 'dispose' ? listener as Disposable : dispose); + return dispose; +} + +export function prependListener(name: K, listener: EventMap[K]) { + getHooks(name).unshift(listener); + const dispose = () => removeListener(name, listener); + _disposables.push(name === 'dispose' ? listener as Disposable : dispose); + return dispose; +} + +export function once(name: K, listener: EventMap[K]) { + // @ts-ignore + const dispose = addListener(name, (...args: Parameters) => { + dispose(); + // @ts-ignore + return listener.apply(this, args); + }); + return dispose; +} + +export function on(name: K, listener: EventMap[K]) { + return addListener(name, listener); +} + +export function off(name: K, listener: EventMap[K]) { + return removeListener(name, listener); +} + +export async function parallel(name: K, ...args: Parameters): Promise { + const tasks: Promise[] = []; + for (const callback of _hooks[name] || []) { + tasks.push(callback.apply(this, args)); + } + await Promise.all(tasks); +} + +export function emit(name: K, ...args: Parameters) { + return parallel(name, ...args); +} + +export async function serial(name: K, ...args: Parameters): Promise> { + for (const callback of _hooks[name] || []) { + const result = await callback.apply(this, args); + if (isBailed(result)) return result; + } + return null; +} + +export function bail(name: K, ...args: Parameters): ReturnType { + for (const callback of _hooks[name] || []) { + const result = callback.apply(this, args); + if (isBailed(result)) return result; + } + return null; +} + +export function boardcast(event: K, ...payload: Parameters) { // Process forked by pm2 would also have process.send - if (isMaster && process.send && !cluster.isMaster) { + if (process.send && !cluster.isMaster) { process.send({ event: 'bus', eventName: event, payload, }); - } else { - if (!bus[event]) bus[event] = {}; - const funcs = Object.keys(bus[event]); - Promise.all(funcs.map((func) => bus[event][func]({ - event, - value: payload, - }))); - } + } else parallel(event, ...payload); } global.Hydro.service.bus = { - subscribe, unsubscribe, publish, + addListener, bail, boardcast, emit, on, off, once, parallel, prependListener, removeListener, serial, }; diff --git a/packages/hydrooj/src/service/db.ts b/packages/hydrooj/src/service/db.ts index 3795c84ef..4bb0ac593 100644 --- a/packages/hydrooj/src/service/db.ts +++ b/packages/hydrooj/src/service/db.ts @@ -21,7 +21,7 @@ mongodb.MongoClient.connect(mongourl, { useNewUrlParser: true, useUnifiedTopolog .then((Client1) => { db2 = Client1.db(opts.name); global.Hydro.service.db.db2 = db2; - bus.publish('system_database_connected', null); + bus.parallel('database/connect', db); }); }); diff --git a/packages/hydrooj/src/service/monitor.ts b/packages/hydrooj/src/service/monitor.ts index cfd5c7507..097ced519 100644 --- a/packages/hydrooj/src/service/monitor.ts +++ b/packages/hydrooj/src/service/monitor.ts @@ -1,5 +1,6 @@ import cluster from 'cluster'; import * as db from './db'; +import * as bus from './bus'; import * as sysinfo from '../lib/sysinfo'; const coll = db.collection('status'); @@ -24,17 +25,15 @@ export function updateJudger(args) { } if (cluster.isMaster) { - global.Hydro.postInit.push( - async () => { - const info = await sysinfo.get(); - await coll.updateOne( - { mid: info.mid, type: 'server' }, - { $set: { ...info, updateAt: new Date(), type: 'server' } }, - { upsert: true }, - ); - setInterval(update, 60 * 1000); - }, - ); + bus.on('app/started', async () => { + const info = await sysinfo.get(); + await coll.updateOne( + { mid: info.mid, type: 'server' }, + { $set: { ...info, updateAt: new Date(), type: 'server' } }, + { upsert: true }, + ); + setInterval(update, 60 * 1000); + }); } global.Hydro.service.monitor = { update, updateJudger }; diff --git a/packages/login-with-qq/package.json b/packages/login-with-qq/package.json index e9552f5fb..027046646 100644 --- a/packages/login-with-qq/package.json +++ b/packages/login-with-qq/package.json @@ -1,6 +1,6 @@ { "name": "@hydrooj/login-with-qq", - "version": "0.0.3", + "version": "0.0.4", "main": "package.json", "repository": "git@github.com:hydro-dev/Hydro.git", "author": "undefined ", @@ -10,7 +10,7 @@ "build": "tsc" }, "peerDependencies": { - "hydrooj": "^2.12.1||^2.13.0" + "hydrooj": "^2.13.1" }, "dependencies": { "ws": "^7.3.1" diff --git a/packages/login-with-qq/service.ts b/packages/login-with-qq/service.ts index 5d8af9dc0..62a9606e9 100644 --- a/packages/login-with-qq/service.ts +++ b/packages/login-with-qq/service.ts @@ -1,4 +1,5 @@ import { } from 'hydrooj'; +import * as bus from 'hydrooj/dist/service/bus'; import WebSocket from 'ws'; export async function postInit() { @@ -32,4 +33,4 @@ export async function postInit() { }); } -global.Hydro.postInit.push(postInit); +bus.subscribe(['app/started'], postInit);