Weishen/Hydro
2
0 Fork 0
mirror of https://gitcode.com/gh_mirrors/hy/Hydro.git synced 2026-10-20 01:38:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

update bus.subscribe

Browse Source
This commit is contained in:
undefined committed 2020-07-30 17:04:46 +08:00
1 parent 875bf696e2
commit d077019e0d
15 files changed
+85 -162
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

hydro/entry/master.ts
+5 -15
View File
@@ -35,7 +35,10 @@ export async function load(call: Entry) {
try {
require('../options').default();
} catch (e) {
await call({ entry: 'setup', newProcess: true });
await call({ entry: 'setup', newProcess: true }).catch((err) => {
console.error('Cannot start setup process.', err);
process.exit(1);
});
}
const bus = require('../service/bus');
await new Promise((resolve) => {
@@ -54,21 +57,8 @@ export async function load(call: Entry) {
const ins = require('../script/upgrade0_1');
await ins.run({ username: 'Root', password: 'rootroot' });
}
for (const i in global.Hydro.service) {
if (global.Hydro.service[i].postInit) {
try {
await global.Hydro.service[i].postInit();
} catch (e) {
console.error(e);
}
}
}
for (const postInit of global.Hydro.postInit) {
try {
await postInit();
} catch (e) {
console.error(e);
}
await postInit().catch(console.error);
}
return await modelSystem.get('server.worker');
}
hydro/handler/home.ts
+21 -8
View File
@@ -206,18 +206,30 @@ class HomeSettingsHandler extends Handler {
this.response.body.settings = setting.PREFERENCE_SETTINGS;
} else if (category === 'account') {
this.response.body.settings = setting.ACCOUNT_SETTINGS;
} else if (category === 'domain') {
this.response.body.settings = setting.DOMAIN_USER_SETTINGS;
} else throw new NotFoundError();
}
async post(args: any) {
const $set = {};
for (const key in args) {
if (setting.SETTINGS_BY_KEY[key]
&& !(setting.SETTINGS_BY_KEY[key].flag & setting.FLAG_DISABLED)) {
$set[key] = args[key];
if (args.category === 'domain') {
for (const key in args) {
if (setting.DOMAIN_USER_SETTINGS_BY_KEY[key]
&& !(setting.DOMAIN_USER_SETTINGS_BY_KEY[key].flag & setting.FLAG_DISABLED)) {
$set[key] = args[key];
}
}
await domain.setUserInDomain(args.domainId, this.user._id, $set);
} else {
for (const key in args) {
if (setting.SETTINGS_BY_KEY[key]
&& !(setting.SETTINGS_BY_KEY[key].flag & setting.FLAG_DISABLED)) {
$set[key] = args[key];
}
}
await user.setById(this.user._id, $set);
}
await user.setById(this.user._id, $set);
this.back();
}
}
@@ -322,7 +334,6 @@ class HomeMessagesHandler extends Handler {
if (!udoc) throw new UserNotFoundError(uid);
if (udoc.gravatar) udoc.gravatar = misc.gravatar(udoc.gravatar);
const mdoc = await message.send(this.user._id, uid, content, message.FLAG_UNREAD);
// TODO(twd2): improve here: projection\
this.back({ mdoc, udoc });
}
@@ -345,8 +356,10 @@ class HomeMessagesHandler extends Handler {
}
class HomeMessagesConnectionHandler extends ConnectionHandler {
id: string;
async prepare() {
bus.subscribe([`user_message-${this.user._id}`], this, 'onMessageReceived');
bus.subscribe([`user_message-${this.user._id}`], this.onMessageReceived.bind(this));
}
async onMessageReceived(e: any) {
@@ -354,7 +367,7 @@ class HomeMessagesConnectionHandler extends ConnectionHandler {
}
async cleanup() {
bus.unsubscribe([`user_message-${this.user._id}`], this, 'onMessageReceived');
bus.unsubscribe([`user_message-${this.user._id}`], this.id);
}
}
hydro/handler/notfound.ts
+1 -3
View File
@@ -8,9 +8,7 @@ class NotFoundHandler extends Handler {
throw new NotFoundError();
}
async get() { }
async post() { }
all() { }
}
export async function apply() {
hydro/handler/problem.ts
+4 -2
View File
@@ -284,6 +284,8 @@ class ProblemPretestConnectionHandler extends ConnectionHandler {
domainId: string;
id: string;
@param('pid', Types.String)
async prepare(domainId: string, pid: string) {
const pdoc = await problem.get(domainId, pid);
@@ -291,7 +293,7 @@ class ProblemPretestConnectionHandler extends ConnectionHandler {
if (!pdoc) throw new ProblemNotFoundError(domainId, pid);
this.pid = pdoc.docId.toString();
this.domainId = domainId;
bus.subscribe(['record_change'], this, 'onRecordChange');
this.id = bus.subscribe(['record_change'], this.onRecordChange.bind(this));
}
async onRecordChange(data) {
@@ -313,7 +315,7 @@ class ProblemPretestConnectionHandler extends ConnectionHandler {
}
async cleanup() {
bus.unsubscribe(['record_change'], this, 'onRecordChange');
bus.unsubscribe(['record_change'], this.id);
}
}
hydro/handler/record.ts
+4 -4
View File
@@ -113,7 +113,7 @@ class RecordMainConnectionHandler extends RecordConnectionHandler {
return;
}
}
bus.subscribe(['record_change'], this, 'onRecordChange');
this.id = bus.subscribe(['record_change'], this.onRecordChange.bind(this));
}
async message(msg) {
@@ -129,7 +129,7 @@ class RecordMainConnectionHandler extends RecordConnectionHandler {
}
async cleanup() {
bus.unsubscribe(['record_change'], this, 'onRecordChange');
bus.unsubscribe(['record_change'], this.id);
}
async onRecordChange(data) {
@@ -157,7 +157,7 @@ class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(Connecti
}
}
this.rid = rid.toString();
bus.subscribe(['record_change'], this, 'onRecordChange');
this.id = bus.subscribe(['record_change'], this.onRecordChange.bind(this));
this.onRecordChange({ value: { rdoc } });
}
@@ -175,7 +175,7 @@ class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(Connecti
}
async cleanup() {
bus.unsubscribe(['record_change'], this, 'onRecordChange');
bus.unsubscribe(['record_change'], this.id);
}
}
hydro/interface.ts
+1 -1
View File
@@ -397,7 +397,7 @@ declare global {
locales: Dict<Dict<string>>,
postInit: Array<() => Promise<any>>,
},
onDestory: Array<() => void>,
onDestory: Array<() => void | Promise<void>>,
addons: string[],
}
}
hydro/model/setting.ts
+17 -13
View File
@@ -7,7 +7,7 @@ import { Setting as _Setting } from '../interface';
type SettingDict = Dictionary<_Setting>;
const countries = moment.tz.countries();
const tzs: Set<string> = new Set();
const tzs = new Set();
for (const country of countries) {
const tz = moment.tz.zonesForCountry(country);
for (const t of tz) tzs.add(t);
@@ -100,6 +100,10 @@ DomainSetting(
Setting('setting_domain', 'gravatar', null, '', 'text', 'gravatar', 'Will be used as the domain icon.'),
Setting('setting_domain', 'bulletin', null, '', 'markdown', 'Bulletin'),
Setting('setting_storage', 'pidCounter', null, 0, 'number', 'Problem ID Counter', null, FLAG_HIDDEN | FLAG_DISABLED),
);
DomainUserSetting(
Setting('setting_info', 'displayName', null, null, 'string', 'display name'),
Setting('setting_storage', 'nAccept', null, 0, 'number', 'nAccept', null, FLAG_HIDDEN | FLAG_DISABLED),
Setting('setting_storage', 'nSubmit', null, 0, 'number', 'nSubmit', null, FLAG_HIDDEN | FLAG_DISABLED),
Setting('setting_storage', 'nLike', null, 0, 'number', 'nLike', null, FLAG_HIDDEN | FLAG_DISABLED),
@@ -121,7 +125,7 @@ SystemSetting(
Setting('setting_server', 'server.cdn', null, '/', 'text', 'CDN Prefix', 'Ends with /'),
Setting('setting_server', 'server.port', null, 8888, 'number', 'Server Port'),
Setting('setting_server', 'server.xff', null, null, 'text', 'IP Header', 'e.g. x-forwarded-for (lowercase)'),
Setting('setting_session', 'session.keys', null, [String.random(32)], 'text', 'session.keys', null, FLAG_DISABLED | FLAG_HIDDEN),
Setting('setting_session', 'session.keys', null, [String.random(32)], 'text', 'session.keys', null, FLAG_HIDDEN),
Setting('setting_session', 'session.secure', null, false, 'boolean', 'session.secure'),
Setting('setting_session', 'session.saved_expire_seconds', null, 3600 * 24 * 30, 'number', 'Saved session expire seconds'),
Setting('setting_session', 'session.unsaved_expire_seconds', null, 3600 * 3, 'number', 'Unsaved session expire seconds'),
@@ -130,17 +134,17 @@ SystemSetting(
Setting('setting_oauth', 'oauth.googleappid', null, null, 'text', 'Google Oauth ClientID', null),
Setting('setting_oauth', 'oauth.googlesecret', null, null, 'text', 'Google Oauth Secret', null, FLAG_SECRET),
Setting('setting_proxy', 'proxy', null, null, 'text', 'Proxy Server URL'),
Setting('setting_constant', 'PROBLEM_PER_PAGE', null, 100, 'number', 'Problems per Page', null, FLAG_HIDDEN),
Setting('setting_constant', 'CONTEST_PER_PAGE', null, 20, 'number', 'Contests per Page', null, FLAG_HIDDEN),
Setting('setting_constant', 'DISCUSSION_PER_PAGE', null, 50, 'number', 'Discussion per Page', null, FLAG_HIDDEN),
Setting('setting_constant', 'RECORD_PER_PAGE', null, 100, 'number', 'Record per Page', null, FLAG_HIDDEN),
Setting('setting_constant', 'SOLUTION_PER_PAGE', null, 20, 'number', 'Solutions per Page', null, FLAG_HIDDEN),
Setting('setting_constant', 'TRAINING_PER_PAGE', null, 10, 'number', 'Training per Page', null, FLAG_HIDDEN),
Setting('setting_constant', 'REPLY_PER_PAGE', null, 50, 'number', 'Reply per Page', null, FLAG_HIDDEN),
Setting('setting_constant', 'HOMEWORK_ON_MAIN', null, 5, 'number', 'HOMEWORK_ON_MAIN', null, FLAG_HIDDEN),
Setting('setting_constant', 'DISCUSSION_ON_MAIN', null, 20, 'number', 'DISCUSSION_ON_MAIN', null, FLAG_HIDDEN),
Setting('setting_constant', 'CONTEST_ON_MAIN', null, 10, 'number', 'CONTEST_ON_MAIN', null, FLAG_HIDDEN),
Setting('setting_constant', 'TRAINING_ON_MAIN', null, 10, 'number', 'TRAINING_ON_MAIN', null, FLAG_HIDDEN),
Setting('setting_storage', 'PROBLEM_PER_PAGE', null, 100, 'number', 'Problems per Page', null, FLAG_HIDDEN),
Setting('setting_storage', 'CONTEST_PER_PAGE', null, 20, 'number', 'Contests per Page', null, FLAG_HIDDEN),
Setting('setting_storage', 'DISCUSSION_PER_PAGE', null, 50, 'number', 'Discussion per Page', null, FLAG_HIDDEN),
Setting('setting_storage', 'RECORD_PER_PAGE', null, 100, 'number', 'Record per Page', null, FLAG_HIDDEN),
Setting('setting_storage', 'SOLUTION_PER_PAGE', null, 20, 'number', 'Solutions per Page', null, FLAG_HIDDEN),
Setting('setting_storage', 'TRAINING_PER_PAGE', null, 10, 'number', 'Training per Page', null, FLAG_HIDDEN),
Setting('setting_storage', 'REPLY_PER_PAGE', null, 50, 'number', 'Reply per Page', null, FLAG_HIDDEN),
Setting('setting_storage', 'HOMEWORK_ON_MAIN', null, 5, 'number', 'HOMEWORK_ON_MAIN', null, FLAG_HIDDEN),
Setting('setting_storage', 'DISCUSSION_ON_MAIN', null, 20, 'number', 'DISCUSSION_ON_MAIN', null, FLAG_HIDDEN),
Setting('setting_storage', 'CONTEST_ON_MAIN', null, 10, 'number', 'CONTEST_ON_MAIN', null, FLAG_HIDDEN),
Setting('setting_storage', 'TRAINING_ON_MAIN', null, 10, 'number', 'TRAINING_ON_MAIN', null, FLAG_HIDDEN),
Setting('setting_storage', 'db.ver', null, 1, 'number', 'Database version', null, FLAG_DISABLED | FLAG_HIDDEN),
Setting('setting_storage', 'user', null, 1, 'number', 'User Counter', null, FLAG_DISABLED | FLAG_HIDDEN),
);
hydro/service/bus.ts
+18 -25
View File
@@ -1,33 +1,24 @@
import { EventEmitter } from 'events';
import cluster from 'cluster';
const bus = new EventEmitter();
const bus = {};
export function subscribe(events, handler, funcName) {
if (!funcName) {
for (const event of events) bus.on(event, handler);
} else {
handler.__bus = (...args) => {
handler[funcName].call(handler, ...args);
};
for (const event of events) bus.on(event, handler.__bus);
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;
}
export function unsubscribe(events: string[], id: string) {
for (const event of events) {
if (!bus[event]) bus[event] = {};
delete bus[event][id];
}
}
export function unsubscribe(events, handler, funcName) {
if (!funcName) {
for (const event of events) bus.off(event, handler);
} else {
handler.__bus = (...args) => {
handler[funcName].call(handler, ...args);
};
// FIXME doesn't work
for (const event of events) bus.off(event, handler.__bus);
delete handler.__bus;
}
}
export function publish(event, payload, isMaster = true) {
export function publish(event: string, payload: any, isMaster = true) {
// Process forked by pm2 would also have process.send
if (isMaster && process.send && !cluster.isMaster) {
process.send({
@@ -36,7 +27,9 @@ export function publish(event, payload, isMaster = true) {
payload,
});
} else {
bus.emit(event, { value: payload, event });
if (!bus[event]) bus[event] = {};
const funcs = Object.keys(bus[event]);
Promise.all(funcs.map((func) => bus[event][func]()));
}
}
hydro/service/server.ts
+9 -34
View File
@@ -532,7 +532,7 @@ async function handle(ctx, HandlerClass, checker) {
} else if (typeof h.post !== 'function') {
throw new MethodNotAllowedError(method);
}
} else if (typeof h[method] !== 'function') {
} else if (typeof h[method] !== 'function' && typeof h.all !== 'function') {
throw new MethodNotAllowedError(method);
}
@@ -623,50 +623,25 @@ export class ConnectionHandler {
return res;
}
async limitRate(op, periodSecs, maxOperations) {
async limitRate(op: string, periodSecs: number, maxOperations: number) {
await opcount.inc(op, this.request.ip, periodSecs, maxOperations);
}
translate(str) {
translate(str: string) {
return str ? str.toString().translate(this.user.viewLang || this.session.viewLang) : '';
}
renderTitle(str) {
renderTitle(str: string) {
return `${this.translate(str)} - Hydro`;
}
checkPerm(...args: Array<bigint[] | bigint>) {
for (const i in args) {
if (args[i] instanceof Array) {
let p = false;
for (const j in args) {
if (this.user.hasPerm(args[i][j])) {
p = true;
break;
}
}
if (!p) throw new PermissionError([args[i]]);
} else if (!this.user.hasPerm(args[i])) {
throw new PermissionError([[args[i]]]);
}
}
checkPerm(...args: bigint[]) {
if (!this.user.hasPerm(...args)) throw new PermissionError(...args);
}
checkPriv(...args: Array<number[] | number>) {
for (const i in args) {
if (args[i] instanceof Array) {
let p = false;
for (const j in args) {
if (this.user.hasPriv(args[i][j])) {
p = true;
break;
}
}
if (!p) throw new PrivilegeError([args[i]]);
} else if (!this.user.hasPriv(args[i])) {
throw new PrivilegeError([[args[i]]]);
}
}
checkPriv(...args: number[]) {
// @ts-ignore
if (!this.user.hasPriv(...args)) throw new PrivilegeError(...args);
}
url(name: string, kwargs = {}) {