mirror of
https://github.com/Tria-plc/emaui.git
synced 2026-08-27 14:21:00 +00:00
718 lines
23 KiB
JavaScript
718 lines
23 KiB
JavaScript
import { a as MF_SERVER_IDENTIFIER, c as WEB_SOCKET_CONNECT_MAGIC_ID, i as DEFAULT_WEB_SOCKET_PORT, l as Message, n as ActionKind, o as UpdateMode } from "./Action-DNNg2YDh.mjs";
|
||
import * as fs$1 from "fs";
|
||
import * as path$1 from "path";
|
||
import { SEPARATOR, createLogger } from "@module-federation/sdk";
|
||
import net from "net";
|
||
import os from "os";
|
||
import WebSocket from "isomorphic-ws";
|
||
import { createServer } from "http";
|
||
import schedule from "node-schedule";
|
||
import { parse } from "url";
|
||
|
||
//#region src/server/message/API/API.ts
|
||
let APIKind = /* @__PURE__ */ function(APIKind) {
|
||
APIKind["UPDATE_SUBSCRIBER"] = "UPDATE_SUBSCRIBER";
|
||
APIKind["RELOAD_WEB_CLIENT"] = "RELOAD_WEB_CLIENT";
|
||
APIKind["FETCH_TYPES"] = "FETCH_TYPES";
|
||
return APIKind;
|
||
}({});
|
||
var API = class extends Message {
|
||
constructor(content, kind) {
|
||
super("API", kind);
|
||
const { code, payload } = content;
|
||
this.code = code;
|
||
this.payload = payload;
|
||
}
|
||
};
|
||
|
||
//#endregion
|
||
//#region src/server/message/API/UpdateSubscriber.ts
|
||
var UpdateSubscriberAPI = class extends API {
|
||
constructor(payload) {
|
||
super({
|
||
code: 0,
|
||
payload
|
||
}, APIKind.UPDATE_SUBSCRIBER);
|
||
}
|
||
};
|
||
|
||
//#endregion
|
||
//#region src/server/message/API/ReloadWebClient.ts
|
||
var ReloadWebClientAPI = class extends API {
|
||
constructor(payload) {
|
||
super({
|
||
code: 0,
|
||
payload
|
||
}, APIKind.RELOAD_WEB_CLIENT);
|
||
}
|
||
};
|
||
|
||
//#endregion
|
||
//#region src/server/message/API/FetchTypes.ts
|
||
var FetchTypesAPI = class extends API {
|
||
constructor(payload) {
|
||
super({
|
||
code: 0,
|
||
payload
|
||
}, APIKind.FETCH_TYPES);
|
||
}
|
||
};
|
||
|
||
//#endregion
|
||
//#region src/server/message/Log/Log.ts
|
||
let LogLevel = /* @__PURE__ */ function(LogLevel) {
|
||
LogLevel["LOG"] = "LOG";
|
||
LogLevel["WARN"] = "WARN";
|
||
LogLevel["ERROR"] = "ERROR";
|
||
return LogLevel;
|
||
}({});
|
||
let LogKind = /* @__PURE__ */ function(LogKind) {
|
||
LogKind["BrokerExitLog"] = "BrokerExitLog";
|
||
LogKind["PublisherRegisteredLog"] = "PublisherRegisteredLog";
|
||
return LogKind;
|
||
}({});
|
||
var Log = class extends Message {
|
||
constructor(level, kind, ignoreVerbose = false) {
|
||
super("Log", kind);
|
||
this.ignoreVerbose = false;
|
||
this.level = level;
|
||
this.ignoreVerbose = ignoreVerbose;
|
||
}
|
||
};
|
||
|
||
//#endregion
|
||
//#region src/server/message/Log/BrokerExitLog.ts
|
||
var BrokerExitLog = class extends Log {
|
||
constructor() {
|
||
super(LogLevel.LOG, LogKind.BrokerExitLog);
|
||
}
|
||
};
|
||
|
||
//#endregion
|
||
//#region src/server/utils/log.ts
|
||
const logger$1 = createLogger(`[ ${MF_SERVER_IDENTIFIER} ]`);
|
||
function fileLog(msg, module, level) {
|
||
if (!process?.env?.["FEDERATION_DEBUG"]) return;
|
||
try {
|
||
const logDir = ".mf";
|
||
const logFile = path$1.join(logDir, "typesGenerate.log");
|
||
const timestamp = (/* @__PURE__ */ new Date()).toISOString();
|
||
if (!fs$1.existsSync(logDir)) fs$1.mkdirSync(logDir, { recursive: true });
|
||
fs$1.appendFileSync(logFile, `[${timestamp}] [${level.toUpperCase()}] ${module} - ${msg}\n`);
|
||
} catch {}
|
||
}
|
||
function error(error, action, from) {
|
||
const err = error instanceof Error ? error : /* @__PURE__ */ new Error(`${action} error`);
|
||
fileLog(`[${action}] error: ${err}`, from, "fatal");
|
||
return err.toString();
|
||
}
|
||
|
||
//#endregion
|
||
//#region src/server/utils/getIPV4.ts
|
||
const localIpv4 = "127.0.0.1";
|
||
const getIpv4Interfaces = () => {
|
||
try {
|
||
const interfaces = os.networkInterfaces();
|
||
const ipv4Interfaces = [];
|
||
Object.values(interfaces).forEach((detail) => {
|
||
detail?.forEach((detail) => {
|
||
const familyV4Value = typeof detail.family === "string" ? "IPv4" : 4;
|
||
if (detail.family === familyV4Value && detail.address !== localIpv4) ipv4Interfaces.push(detail);
|
||
});
|
||
});
|
||
return ipv4Interfaces;
|
||
} catch (_err) {
|
||
return [];
|
||
}
|
||
};
|
||
const getIPV4 = () => {
|
||
return (getIpv4Interfaces()[0] || { address: localIpv4 }).address;
|
||
};
|
||
|
||
//#endregion
|
||
//#region src/server/utils/index.ts
|
||
function getIdentifier(options) {
|
||
const { ip, name } = options;
|
||
return `mf ${SEPARATOR}${name}${ip ? `${SEPARATOR}${ip}` : ""}`;
|
||
}
|
||
function fib(n) {
|
||
let i = 2;
|
||
const res = [
|
||
0,
|
||
1,
|
||
1
|
||
];
|
||
while (i <= n) {
|
||
res[i] = res[i - 1] + res[i - 2];
|
||
i++;
|
||
}
|
||
return res[n];
|
||
}
|
||
function getFreePort() {
|
||
return new Promise((resolve, reject) => {
|
||
const server = net.createServer();
|
||
server.unref();
|
||
server.on("error", reject);
|
||
server.listen(0, () => {
|
||
const { port } = server.address();
|
||
server.close(() => {
|
||
resolve(port);
|
||
});
|
||
});
|
||
});
|
||
}
|
||
|
||
//#endregion
|
||
//#region src/server/Publisher.ts
|
||
var Publisher = class {
|
||
constructor(ctx) {
|
||
this._name = ctx.name;
|
||
this._ip = ctx.ip;
|
||
this._remoteTypeTarPath = ctx.remoteTypeTarPath;
|
||
this._subscribers = /* @__PURE__ */ new Map();
|
||
this._ws = ctx.ws;
|
||
this.dynamicRemoteMap = /* @__PURE__ */ new Map();
|
||
}
|
||
get identifier() {
|
||
return getIdentifier({
|
||
name: this._name,
|
||
ip: this._ip
|
||
});
|
||
}
|
||
get name() {
|
||
return this._name;
|
||
}
|
||
get ip() {
|
||
return this._ip;
|
||
}
|
||
get remoteTypeTarPath() {
|
||
return this._remoteTypeTarPath;
|
||
}
|
||
get hasSubscribes() {
|
||
return Boolean(this._subscribers.size);
|
||
}
|
||
get subscribers() {
|
||
return this._subscribers;
|
||
}
|
||
addSubscriber(identifier, subscriber) {
|
||
fileLog(`${this.name} set subscriber: ${identifier}`, "Publisher", "info");
|
||
this._subscribers.set(identifier, subscriber);
|
||
}
|
||
removeSubscriber(identifier) {
|
||
if (this._subscribers.has(identifier)) {
|
||
fileLog(`${this.name} removeSubscriber: ${identifier}`, "Publisher", "warn");
|
||
this._subscribers.delete(identifier);
|
||
}
|
||
}
|
||
notifySubscriber(subscriberIdentifier, options) {
|
||
const subscriber = this._subscribers.get(subscriberIdentifier);
|
||
if (!subscriber) {
|
||
fileLog(`[notifySubscriber] ${this.name} notifySubscriber: ${subscriberIdentifier}, does not exits`, "Publisher", "error");
|
||
return;
|
||
}
|
||
const api = new UpdateSubscriberAPI(options);
|
||
subscriber.send(JSON.stringify(api));
|
||
fileLog(`[notifySubscriber] ${this.name} notifySubscriber: ${JSON.stringify(subscriberIdentifier)}, message: ${JSON.stringify(api)}`, "Publisher", "info");
|
||
}
|
||
fetchRemoteTypes(options) {
|
||
fileLog(`[fetchRemoteTypes] ${this.name} fetchRemoteTypes, options: ${JSON.stringify(options)}, ws: ${Boolean(this._ws)}`, "Publisher", "info");
|
||
if (!this._ws) return;
|
||
const api = new FetchTypesAPI(options);
|
||
this._ws.send(JSON.stringify(api));
|
||
}
|
||
notifySubscribers(options) {
|
||
const api = new UpdateSubscriberAPI(options);
|
||
this.broadcast(api);
|
||
}
|
||
broadcast(message) {
|
||
if (this.hasSubscribes) this._subscribers.forEach((subscriber, key) => {
|
||
fileLog(`[BroadCast] ${this.name} notifySubscriber: ${key}, PID: ${process.pid}, message: ${JSON.stringify(message)}`, "Publisher", "info");
|
||
subscriber.send(JSON.stringify(message));
|
||
});
|
||
else fileLog(`[BroadCast] ${this.name}'s subscribe is empty`, "Publisher", "warn");
|
||
}
|
||
close() {
|
||
this._ws = void 0;
|
||
this._subscribers.forEach((_subscriber, identifier) => {
|
||
fileLog(`[BroadCast] close ${this.name} remove: ${identifier}`, "Publisher", "warn");
|
||
this.removeSubscriber(identifier);
|
||
});
|
||
}
|
||
};
|
||
|
||
//#endregion
|
||
//#region src/server/message/Action/Update.ts
|
||
let UpdateKind = /* @__PURE__ */ function(UpdateKind) {
|
||
UpdateKind["UPDATE_TYPE"] = "UPDATE_TYPE";
|
||
UpdateKind["RELOAD_PAGE"] = "RELOAD_PAGE";
|
||
return UpdateKind;
|
||
}({});
|
||
|
||
//#endregion
|
||
//#region src/server/broker/Broker.ts
|
||
var Broker = class Broker {
|
||
static {
|
||
this.WEB_SOCKET_CONNECT_MAGIC_ID = WEB_SOCKET_CONNECT_MAGIC_ID;
|
||
}
|
||
static {
|
||
this.DEFAULT_WEB_SOCKET_PORT = DEFAULT_WEB_SOCKET_PORT;
|
||
}
|
||
static {
|
||
this.DEFAULT_SECURE_WEB_SOCKET_PORT = 16324;
|
||
}
|
||
static {
|
||
this.DEFAULT_WAITING_TIME = 1.5 * 60 * 60 * 1e3;
|
||
}
|
||
constructor() {
|
||
this._publisherMap = /* @__PURE__ */ new Map();
|
||
this._webClientMap = /* @__PURE__ */ new Map();
|
||
this._tmpSubscriberShelter = /* @__PURE__ */ new Map();
|
||
this._scheduleJob = null;
|
||
this._setSchedule();
|
||
this._startWsServer();
|
||
this._stopWhenSIGTERMOrSIGINT();
|
||
this._handleUnexpectedExit();
|
||
}
|
||
get hasPublishers() {
|
||
return Boolean(this._publisherMap.size);
|
||
}
|
||
async _startWsServer() {
|
||
const wsHandler = (ws, req) => {
|
||
const { url: reqUrl = "" } = req;
|
||
const { query } = parse(reqUrl, true);
|
||
const { WEB_SOCKET_CONNECT_MAGIC_ID } = query;
|
||
if (WEB_SOCKET_CONNECT_MAGIC_ID === Broker.WEB_SOCKET_CONNECT_MAGIC_ID) {
|
||
ws.on("message", (message) => {
|
||
try {
|
||
const text = message.toString();
|
||
const action = JSON.parse(text);
|
||
fileLog(`${action?.kind} action received `, "Broker", "info");
|
||
this._takeAction(action, ws);
|
||
} catch (error) {
|
||
fileLog(`parse action message error: ${error}`, "Broker", "error");
|
||
}
|
||
});
|
||
ws.on("error", (e) => {
|
||
fileLog(`parse action message error: ${e}`, "Broker", "error");
|
||
});
|
||
} else {
|
||
ws.send("Invalid CONNECT ID.");
|
||
fileLog("Invalid CONNECT ID.", "Broker", "warn");
|
||
ws.close();
|
||
}
|
||
};
|
||
const server = createServer();
|
||
this._webSocketServer = new WebSocket.Server({ noServer: true });
|
||
this._webSocketServer.on("error", (err) => {
|
||
fileLog(`ws error: \n${err.message}\n ${err.stack}`, "Broker", "error");
|
||
});
|
||
this._webSocketServer.on("listening", () => {
|
||
fileLog(`WebSocket server is listening on port ${Broker.DEFAULT_WEB_SOCKET_PORT}`, "Broker", "info");
|
||
});
|
||
this._webSocketServer.on("connection", wsHandler);
|
||
this._webSocketServer.on("close", (code) => {
|
||
fileLog(`WebSocket Server Close with Code ${code}`, "Broker", "warn");
|
||
this._webSocketServer && this._webSocketServer.close();
|
||
this._webSocketServer = void 0;
|
||
});
|
||
server.on("upgrade", (req, socket, head) => {
|
||
if (req.url) {
|
||
const { pathname } = parse(req.url);
|
||
if (pathname === "/") this._webSocketServer?.handleUpgrade(req, socket, head, (ws) => {
|
||
this._webSocketServer?.emit("connection", ws, req);
|
||
});
|
||
}
|
||
});
|
||
server.listen(Broker.DEFAULT_WEB_SOCKET_PORT);
|
||
}
|
||
async _takeAction(action, client) {
|
||
const { kind, payload } = action;
|
||
if (kind === ActionKind.ADD_PUBLISHER) await this._addPublisher(payload, client);
|
||
if (kind === ActionKind.UPDATE_PUBLISHER) await this._updatePublisher(payload, client);
|
||
if (kind === ActionKind.ADD_SUBSCRIBER) await this._addSubscriber(payload, client);
|
||
if (kind === ActionKind.EXIT_SUBSCRIBER) await this._removeSubscriber(payload, client);
|
||
if (kind === ActionKind.EXIT_PUBLISHER) await this._removePublisher(payload, client);
|
||
if (kind === ActionKind.ADD_WEB_CLIENT) await this._addWebClient(payload, client);
|
||
if (kind === ActionKind.NOTIFY_WEB_CLIENT) await this._notifyWebClient(payload, client);
|
||
if (kind === ActionKind.FETCH_TYPES) await this._fetchTypes(payload, client);
|
||
if (kind === ActionKind.ADD_DYNAMIC_REMOTE) this._addDynamicRemote(payload);
|
||
}
|
||
async _addPublisher(context, client) {
|
||
const { name, ip, remoteTypeTarPath } = context ?? {};
|
||
const identifier = getIdentifier({
|
||
name,
|
||
ip
|
||
});
|
||
if (this._publisherMap.has(identifier)) {
|
||
fileLog(`[${ActionKind.ADD_PUBLISHER}] ${identifier} has been added, this action will be ignored`, "Broker", "warn");
|
||
return;
|
||
}
|
||
try {
|
||
const publisher = new Publisher({
|
||
name,
|
||
ip,
|
||
remoteTypeTarPath,
|
||
ws: client
|
||
});
|
||
this._publisherMap.set(identifier, publisher);
|
||
fileLog(`[${ActionKind.ADD_PUBLISHER}] ${identifier} Adding Publisher Succeed`, "Broker", "info");
|
||
const tmpSubScribers = this._getTmpSubScribers(identifier);
|
||
if (tmpSubScribers) {
|
||
fileLog(`[${ActionKind.ADD_PUBLISHER}] consumeTmpSubscriber set ${publisher.name}’s subscribers `, "Broker", "info");
|
||
this._consumeTmpSubScribers(publisher, tmpSubScribers);
|
||
this._clearTmpSubScriberRelation(identifier);
|
||
}
|
||
} catch (err) {
|
||
const msg = error(err, ActionKind.ADD_PUBLISHER, "Broker");
|
||
client.send(msg);
|
||
client.close();
|
||
}
|
||
}
|
||
async _updatePublisher(context, client) {
|
||
const { name, updateMode, updateKind, updateSourcePaths, remoteTypeTarPath, ip } = context ?? {};
|
||
const identifier = getIdentifier({
|
||
name,
|
||
ip
|
||
});
|
||
if (!this._publisherMap.has(identifier)) {
|
||
fileLog(`[${ActionKind.UPDATE_PUBLISHER}] ${identifier} has not been started, this action will be ignored
|
||
this._publisherMap: ${JSON.stringify(this._publisherMap.entries())}
|
||
`, "Broker", "warn");
|
||
return;
|
||
}
|
||
try {
|
||
const publisher = this._publisherMap.get(identifier);
|
||
fileLog(`[${ActionKind.UPDATE_PUBLISHER}] ${identifier} update, and notify subscribers to update`, "Broker", "info");
|
||
if (publisher) {
|
||
publisher.notifySubscribers({
|
||
remoteTypeTarPath,
|
||
name,
|
||
updateMode,
|
||
updateKind,
|
||
updateSourcePaths: updateSourcePaths || []
|
||
});
|
||
this._publisherMap.forEach((p) => {
|
||
if (p.name === publisher.name) return;
|
||
const dynamicRemoteInfo = p.dynamicRemoteMap.get(identifier);
|
||
if (dynamicRemoteInfo) {
|
||
fileLog(`dynamicRemoteInfo: ${JSON.stringify(dynamicRemoteInfo)}, identifier:${identifier} publish: ${p.name}`, "Broker", "info");
|
||
p.fetchRemoteTypes({
|
||
remoteInfo: dynamicRemoteInfo,
|
||
once: false
|
||
});
|
||
}
|
||
});
|
||
}
|
||
} catch (err) {
|
||
const msg = error(err, ActionKind.UPDATE_PUBLISHER, "Broker");
|
||
client.send(msg);
|
||
client.close();
|
||
}
|
||
}
|
||
async _fetchTypes(context, _client) {
|
||
const { name, ip, remoteInfo } = context ?? {};
|
||
const identifier = getIdentifier({
|
||
name,
|
||
ip
|
||
});
|
||
try {
|
||
const publisher = this._publisherMap.get(identifier);
|
||
fileLog(`[${ActionKind.FETCH_TYPES}] ${identifier} fetch types`, "Broker", "info");
|
||
if (publisher) publisher.fetchRemoteTypes({
|
||
remoteInfo,
|
||
once: true
|
||
});
|
||
} catch (err) {
|
||
fileLog(`[${ActionKind.FETCH_TYPES}] ${identifier} fetch types fail , error info: ${err}`, "Broker", "error");
|
||
}
|
||
}
|
||
_addDynamicRemote(context) {
|
||
const { name, ip, remoteInfo, remoteIp } = context ?? {};
|
||
const identifier = getIdentifier({
|
||
name,
|
||
ip
|
||
});
|
||
const publisher = this._publisherMap.get(identifier);
|
||
const remoteId = getIdentifier({
|
||
name: remoteInfo.name,
|
||
ip: remoteIp
|
||
});
|
||
fileLog(`[${ActionKind.ADD_DYNAMIC_REMOTE}] identifier:${identifier},publisher: ${publisher.name}, remoteId:${remoteId}`, "Broker", "error");
|
||
if (!publisher || publisher.dynamicRemoteMap.has(remoteId)) return;
|
||
publisher.dynamicRemoteMap.set(remoteId, remoteInfo);
|
||
}
|
||
async _addSubscriber(context, client) {
|
||
const { publishers, name: subscriberName } = context ?? {};
|
||
publishers.forEach((publisher) => {
|
||
const { name, ip } = publisher;
|
||
const identifier = getIdentifier({
|
||
name,
|
||
ip
|
||
});
|
||
if (!this._publisherMap.has(identifier)) {
|
||
fileLog(`[${ActionKind.ADD_SUBSCRIBER}]: ${identifier} has not been started, ${subscriberName} will add the relation to tmp shelter`, "Broker", "warn");
|
||
this._addTmpSubScriberRelation({
|
||
name: getIdentifier({
|
||
name: context.name,
|
||
ip: context.ip
|
||
}),
|
||
client
|
||
}, publisher);
|
||
return;
|
||
}
|
||
try {
|
||
const registeredPublisher = this._publisherMap.get(identifier);
|
||
if (registeredPublisher) {
|
||
registeredPublisher.addSubscriber(getIdentifier({
|
||
name: subscriberName,
|
||
ip: context.ip
|
||
}), client);
|
||
fileLog(`[${ActionKind.ADD_SUBSCRIBER}]: ${identifier} has been started, Adding Subscriber ${subscriberName} Succeed, this.__publisherMap are: ${JSON.stringify(Array.from(this._publisherMap.entries()))}`, "Broker", "info");
|
||
registeredPublisher.notifySubscriber(getIdentifier({
|
||
name: subscriberName,
|
||
ip: context.ip
|
||
}), {
|
||
updateKind: UpdateKind.UPDATE_TYPE,
|
||
updateMode: UpdateMode.PASSIVE,
|
||
updateSourcePaths: [registeredPublisher.name],
|
||
remoteTypeTarPath: registeredPublisher.remoteTypeTarPath,
|
||
name: registeredPublisher.name
|
||
});
|
||
fileLog(`[${ActionKind.ADD_SUBSCRIBER}]: notifySubscriber Subscriber ${subscriberName}, updateMode: "PASSIVE", updateSourcePaths: ${registeredPublisher.name}`, "Broker", "info");
|
||
}
|
||
} catch (err) {
|
||
const msg = error(err, ActionKind.ADD_SUBSCRIBER, "Broker");
|
||
client.send(msg);
|
||
client.close();
|
||
}
|
||
});
|
||
}
|
||
async _removeSubscriber(context, client) {
|
||
const { publishers } = context ?? {};
|
||
const subscriberIdentifier = getIdentifier({
|
||
name: context?.name,
|
||
ip: context?.ip
|
||
});
|
||
publishers.forEach((publisher) => {
|
||
const { name, ip } = publisher;
|
||
const identifier = getIdentifier({
|
||
name,
|
||
ip
|
||
});
|
||
const registeredPublisher = this._publisherMap.get(identifier);
|
||
if (!registeredPublisher) {
|
||
fileLog(`[${ActionKind.EXIT_SUBSCRIBER}], ${identifier} does not exit `, "Broker", "warn");
|
||
return;
|
||
}
|
||
try {
|
||
fileLog(`[${ActionKind.EXIT_SUBSCRIBER}], ${identifier} will exit `, "Broker", "INFO");
|
||
registeredPublisher.removeSubscriber(subscriberIdentifier);
|
||
this._clearTmpSubScriberRelation(identifier);
|
||
if (!registeredPublisher.hasSubscribes) this._publisherMap.delete(identifier);
|
||
if (!this.hasPublishers) this.exit();
|
||
} catch (err) {
|
||
const msg = error(err, ActionKind.EXIT_SUBSCRIBER, "Broker");
|
||
client.send(msg);
|
||
client.close();
|
||
}
|
||
});
|
||
}
|
||
async _removePublisher(context, client) {
|
||
const { name, ip } = context ?? {};
|
||
const identifier = getIdentifier({
|
||
name,
|
||
ip
|
||
});
|
||
const publisher = this._publisherMap.get(identifier);
|
||
if (!publisher) {
|
||
fileLog(`[${ActionKind.EXIT_PUBLISHER}]: ${identifier}} has not been added, this action will be ingored`, "Broker", "warn");
|
||
return;
|
||
}
|
||
try {
|
||
const { subscribers } = publisher;
|
||
subscribers.forEach((subscriber, subscriberIdentifier) => {
|
||
this._addTmpSubScriberRelation({
|
||
name: subscriberIdentifier,
|
||
client: subscriber
|
||
}, {
|
||
name: publisher.name,
|
||
ip: publisher.ip
|
||
});
|
||
fileLog(`[${ActionKind.EXIT_PUBLISHER}]: ${identifier} is removing , subscriber: ${subscriberIdentifier} will be add tmpSubScriberRelation`, "Broker", "info");
|
||
});
|
||
this._publisherMap.delete(identifier);
|
||
fileLog(`[${ActionKind.EXIT_PUBLISHER}]: ${identifier} is removed `, "Broker", "info");
|
||
if (!this.hasPublishers) {
|
||
fileLog(`[${ActionKind.EXIT_PUBLISHER}]: _publisherMap is empty, all server will exit `, "Broker", "warn");
|
||
this.exit();
|
||
}
|
||
} catch (err) {
|
||
const msg = error(err, ActionKind.EXIT_PUBLISHER, "Broker");
|
||
client.send(msg);
|
||
client.close();
|
||
}
|
||
}
|
||
async _addWebClient(context, client) {
|
||
const { name } = context ?? {};
|
||
const identifier = getIdentifier({ name });
|
||
if (this._webClientMap.has(identifier)) fileLog(`${identifier}} has been added, this action will override prev WebClient`, "Broker", "warn");
|
||
try {
|
||
this._webClientMap.set(identifier, client);
|
||
fileLog(`${identifier} adding WebClient Succeed`, "Broker", "info");
|
||
} catch (err) {
|
||
const msg = error(err, ActionKind.ADD_WEB_CLIENT, "Broker");
|
||
client.send(msg);
|
||
client.close();
|
||
}
|
||
}
|
||
async _notifyWebClient(context, client) {
|
||
const { name, updateMode } = context ?? {};
|
||
const identifier = getIdentifier({ name });
|
||
const webClient = this._webClientMap.get(identifier);
|
||
if (!webClient) {
|
||
fileLog(`[${ActionKind.NOTIFY_WEB_CLIENT}] ${identifier} has not been added, this action will be ignored`, "Broker", "warn");
|
||
return;
|
||
}
|
||
try {
|
||
const api = new ReloadWebClientAPI({
|
||
name,
|
||
updateMode
|
||
});
|
||
webClient.send(JSON.stringify(api));
|
||
fileLog(`[${ActionKind.NOTIFY_WEB_CLIENT}] Notify ${name} WebClient Succeed`, "Broker", "info");
|
||
} catch (err) {
|
||
const msg = error(err, ActionKind.NOTIFY_WEB_CLIENT, "Broker");
|
||
client.send(msg);
|
||
client.close();
|
||
}
|
||
}
|
||
_addTmpSubScriberRelation(subscriber, publisher) {
|
||
const publisherIdentifier = getIdentifier({
|
||
name: publisher.name,
|
||
ip: publisher.ip
|
||
});
|
||
const subscriberIdentifier = subscriber.name;
|
||
const shelter = this._tmpSubscriberShelter.get(publisherIdentifier);
|
||
if (!shelter) {
|
||
const map = /* @__PURE__ */ new Map();
|
||
map.set(subscriberIdentifier, subscriber);
|
||
this._tmpSubscriberShelter.set(publisherIdentifier, {
|
||
subscribers: map,
|
||
timestamp: Date.now()
|
||
});
|
||
fileLog(`[AddTmpSubscriberRelation] ${publisherIdentifier}'s subscriber has ${subscriberIdentifier} `, "Broker", "info");
|
||
return;
|
||
}
|
||
if (shelter.subscribers.get(subscriberIdentifier)) {
|
||
fileLog(`[AddTmpSubscriberRelation] ${publisherIdentifier} and ${subscriberIdentifier} relation has been added`, "Broker", "warn");
|
||
shelter.subscribers.set(subscriberIdentifier, subscriber);
|
||
shelter.timestamp = Date.now();
|
||
} else {
|
||
fileLog(`AddTmpSubscriberLog ${publisherIdentifier}'s shelter has been added, update shelter.subscribers ${subscriberIdentifier}`, "Broker", "warn");
|
||
shelter.subscribers.set(subscriberIdentifier, subscriber);
|
||
}
|
||
}
|
||
_getTmpSubScribers(publisherIdentifier) {
|
||
return this._tmpSubscriberShelter.get(publisherIdentifier)?.subscribers;
|
||
}
|
||
_consumeTmpSubScribers(publisher, tmpSubScribers) {
|
||
tmpSubScribers.forEach((tmpSubScriber, identifier) => {
|
||
fileLog(`notifyTmpSubScribers ${publisher.name} will be add a subscriber: ${identifier} `, "Broker", "warn");
|
||
publisher.addSubscriber(identifier, tmpSubScriber.client);
|
||
publisher.notifySubscriber(identifier, {
|
||
updateKind: UpdateKind.UPDATE_TYPE,
|
||
updateMode: UpdateMode.PASSIVE,
|
||
updateSourcePaths: [publisher.name],
|
||
remoteTypeTarPath: publisher.remoteTypeTarPath,
|
||
name: publisher.name
|
||
});
|
||
});
|
||
}
|
||
_clearTmpSubScriberRelation(identifier) {
|
||
this._tmpSubscriberShelter.delete(identifier);
|
||
}
|
||
_clearTmpSubScriberRelations() {
|
||
this._tmpSubscriberShelter.clear();
|
||
}
|
||
_disconnect() {
|
||
this._publisherMap.forEach((publisher) => {
|
||
publisher.close();
|
||
});
|
||
}
|
||
_setSchedule() {
|
||
const rule = new schedule.RecurrenceRule();
|
||
if (Number(process.env["FEDERATION_SERVER_TEST"])) {
|
||
const interval = Number(process.env["FEDERATION_SERVER_TEST"]) / 1e3;
|
||
const second = [];
|
||
for (let i = 0; i < 60; i = i + interval) second.push(i);
|
||
rule.second = second;
|
||
} else {
|
||
rule.second = 0;
|
||
rule.hour = [
|
||
0,
|
||
3,
|
||
6,
|
||
9,
|
||
12,
|
||
15,
|
||
18
|
||
];
|
||
rule.minute = 0;
|
||
}
|
||
const serverTest = Number(process.env["FEDERATION_SERVER_TEST"]);
|
||
this._scheduleJob = schedule.scheduleJob(rule, () => {
|
||
this._tmpSubscriberShelter.forEach((tmpSubscriber, identifier) => {
|
||
fileLog(` _clearTmpSubScriberRelation ${identifier}, ${Date.now() - tmpSubscriber.timestamp >= (process.env["GARFISH_MODULE_SERVER_TEST"] ? serverTest : Broker.DEFAULT_WAITING_TIME)}`, "Broker", "info");
|
||
if (Date.now() - tmpSubscriber.timestamp >= (process.env["FEDERATION_SERVER_TEST"] ? serverTest : Broker.DEFAULT_WAITING_TIME)) this._clearTmpSubScriberRelation(identifier);
|
||
});
|
||
});
|
||
}
|
||
_clearSchedule() {
|
||
if (!this._scheduleJob) return;
|
||
this._scheduleJob.cancel();
|
||
this._scheduleJob = null;
|
||
}
|
||
_stopWhenSIGTERMOrSIGINT() {
|
||
process.on("SIGTERM", () => {
|
||
this.exit();
|
||
});
|
||
process.on("SIGINT", () => {
|
||
this.exit();
|
||
});
|
||
}
|
||
_handleUnexpectedExit() {
|
||
process.on("unhandledRejection", (error) => {
|
||
console.error("Unhandled Rejection Error: ", error);
|
||
fileLog(`Unhandled Rejection Error: ${error}`, "Broker", "fatal");
|
||
process.exit(1);
|
||
});
|
||
process.on("uncaughtException", (error) => {
|
||
console.error("Unhandled Exception Error: ", error);
|
||
fileLog(`Unhandled Rejection Error: ${error}`, "Broker", "fatal");
|
||
process.exit(1);
|
||
});
|
||
}
|
||
async start() {}
|
||
exit() {
|
||
const brokerExitLog = new BrokerExitLog();
|
||
this.broadcast(JSON.stringify(brokerExitLog));
|
||
this._disconnect();
|
||
this._clearSchedule();
|
||
this._clearTmpSubScriberRelations();
|
||
this._webSocketServer && this._webSocketServer.close();
|
||
this._secureWebSocketServer && this._secureWebSocketServer.close();
|
||
process.exit(0);
|
||
}
|
||
broadcast(message) {
|
||
fileLog(`[broadcast] exit info : ${JSON.stringify(message)}`, "Broker", "warn");
|
||
this._webSocketServer?.clients.forEach((client) => {
|
||
client.send(JSON.stringify(message));
|
||
});
|
||
this._secureWebSocketServer?.clients.forEach((client) => {
|
||
client.send(JSON.stringify(message));
|
||
});
|
||
}
|
||
};
|
||
|
||
//#endregion
|
||
export { getIdentifier as a, logger$1 as c, getFreePort as i, LogKind as l, UpdateKind as n, getIPV4 as o, fib as r, fileLog as s, Broker as t, APIKind as u }; |