core: bug fixes

This commit is contained in:
undefined committed 2022-10-06 02:53:22 +08:00
1 parent 5d340608db
commit 2daa7e2034
11 files changed
+36 -34

No files matched your search

+2 -1
View File
@@ -9,6 +9,7 @@ import { Context } from '../context';
import i18n from '../lib/i18n';
import { Logger } from '../logger';
import * as bus from '../service/bus';
import { unwrapExports } from '../utils';
const logger = new Logger('common');
@@ -37,7 +38,7 @@ const getLoader = (type: LoadTask, filename: string) => async function loader(pe
const p = locateFile(i, [`${filename}.ts`, `${filename}.js`]);
if (p && !fail.includes(i)) {
try {
const m = require(p);
const m = unwrapExports(require(p));
if (m.apply) ctx.loader.reloadPlugin(ctx, p, {});
else logger.info(`${type.replace(/^(.)/, (t) => t.toUpperCase())} init: %s`, i);
} catch (e) {
+1 -1
View File
@@ -60,6 +60,7 @@ export async function apply(ctx: Context) {
}
await handler(pending, fail, ctx);
await addon(pending, fail, ctx);
await ctx.lifecycle.flush();
for (const i in global.Hydro.handler) await global.Hydro.handler[i]();
const scriptDir = path.resolve(__dirname, '..', 'script');
for (const h of await fs.readdir(scriptDir)) {
@@ -68,7 +69,6 @@ export async function apply(ctx: Context) {
await ctx.lifecycle.flush();
await script(pending, fail, ctx);
await ctx.lifecycle.flush();
await sleep(1000); // lifecycle.flush won't wait for async function
await ctx.parallel('app/started');
if (process.env.NODE_APP_INSTANCE === '0') {
const scripts = require('../upgrade').default;
+1 -1
View File
@@ -556,7 +556,7 @@ export interface Task {
}
export interface FileNode {
/** File Path In MinIO */
/** File Path In S3 */
_id: string;
/** Actual File Path */
path: string;
+3 -3
View File
@@ -54,12 +54,12 @@ export class Loader {
async reloadPlugin(parent: Context, key: string, config: any, asName = '') {
let fork = parent.state[Loader.Record]?.[key];
if (fork) {
logger.info('reload plugin %c', key);
logger.info('reload plugin %c', key.split('node_modules').pop());
fork.update(config);
} else {
logger.info('apply plugin %c', key);
logger.info('apply plugin %c', key.split('node_modules').pop());
const plugin = await this.resolvePlugin(key);
if (!plugin) return;
if (!plugin) return;
resolveConfig(plugin, config);
if (asName) plugin.name = asName;
// fork = parent.plugin(plugin, this.interpolate(config));
+1 -1
View File
@@ -186,7 +186,7 @@ export async function apply(ctx: Context) {
const stream = collEvent.watch();
const handleEvent = async (doc: EventDoc) => {
const payload = JSON.parse(doc.payload);
if (process.send) process.send({ type: 'hydro:broadcast', data: { event: doc.event, payload } });
process.send?.({ type: 'hydro:broadcast', data: { event: doc.event, payload } });
await (bus.parallel as any)(doc.event, ...payload);
};
stream.on('change', async (change) => {
+14 -17
View File
@@ -3,7 +3,6 @@ import type {
Db, FilterQuery, ObjectID, OnlyFieldsOfType,
} from 'mongodb';
import pm2 from '@hydrooj/utils/lib/locate-pm2';
import { Context } from '../context';
import type { ProblemSolutionHandler } from '../handler/problem';
import type { UserRegisterHandler } from '../handler/user';
import type {
@@ -100,23 +99,21 @@ export interface EventMap extends LifecycleEvents, HandlerEvents {
}
/* eslint-enable @typescript-eslint/naming-convention */
export function apply(ctx: Context) {
try {
if (!process.send || !pm2 || process.env.exec_mode !== 'cluster_mode') throw new Error();
pm2.launchBus((err, bus) => {
if (err) throw new Error();
bus.on('hydro:broadcast', (packet) => {
(ctx.parallel as any)(packet.data.event, ...packet.data.payload);
});
ctx.on('bus/broadcast', (event, payload) => {
process.send({ type: 'hydro:broadcast', data: { event, payload } });
});
console.debug('Using pm2 event bus');
try {
if (!process.send || !pm2 || process.env.exec_mode !== 'cluster_mode') throw new Error();
pm2.launchBus((err, bus) => {
if (err) throw new Error();
bus.on('hydro:broadcast', (packet) => {
(app.parallel as any)(packet.data.event, ...packet.data.payload);
});
} catch (e) {
ctx.on('bus/broadcast', (event, payload) => ctx.parallel(event, ...payload));
console.debug('Using mongodb external event bus');
}
app.on('bus/broadcast', (event, payload) => {
process.send({ type: 'hydro:broadcast', data: { event, payload } });
});
console.debug('Using pm2 event bus');
});
} catch (e) {
app.on('bus/broadcast', (event, payload) => app.parallel(event, ...payload));
console.debug('Using mongodb external event bus');
}
export default app;
+3 -3
View File
@@ -209,8 +209,8 @@ export class Handler extends HandlerCommon {
}
}
async function bail(name: string, ...args: any[]) {
const r = await (bus.bail as any)(name, ...args);
async function serial(name: string, ...args: any[]) {
const r = await (bus.serial as any)(name, ...args);
if (r instanceof Error) throw r;
return r;
}
@@ -263,7 +263,7 @@ async function handle(ctx: KoaContext, HandlerClass, checker) {
let control;
if (step.startsWith('log/')) h.args[step.slice(4)] = Date.now();
// eslint-disable-next-line no-await-in-loop
else if (step.startsWith('handler/')) control = await bail(step, h);
else if (step.startsWith('handler/')) control = await serial(step, h);
// eslint-disable-next-line no-await-in-loop
else if (typeof h[step] === 'function') control = await h[step](args);
if (control) {