aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/server/DashSession/DashSessionAgent.ts5
-rw-r--r--src/server/DashSession/Session/agents/applied_session_agent.ts58
-rw-r--r--src/server/DashSession/Session/agents/monitor.ts298
-rw-r--r--src/server/DashSession/Session/agents/process_message_router.ts41
-rw-r--r--src/server/DashSession/Session/agents/promisified_ipc_manager.ts173
-rw-r--r--src/server/DashSession/Session/agents/server_worker.ts160
-rw-r--r--src/server/DashSession/Session/utilities/repl.ts128
-rw-r--r--src/server/DashSession/Session/utilities/session_config.ts129
-rw-r--r--src/server/DashSession/Session/utilities/utilities.ts37
-rw-r--r--src/server/index.ts2
10 files changed, 1029 insertions, 2 deletions
diff --git a/src/server/DashSession/DashSessionAgent.ts b/src/server/DashSession/DashSessionAgent.ts
index d61e9aac1..85bfe14de 100644
--- a/src/server/DashSession/DashSessionAgent.ts
+++ b/src/server/DashSession/DashSessionAgent.ts
@@ -8,8 +8,11 @@ import { launchServer, onWindows } from "..";
import { readdirSync, statSync, createWriteStream, readFileSync, unlinkSync } from "fs";
import * as Archiver from "archiver";
import { resolve } from "path";
-import { AppliedSessionAgent, MessageHandler, ExitHandler, Monitor, ServerWorker } from "resilient-server-session";
import rimraf = require("rimraf");
+import { AppliedSessionAgent, ExitHandler } from "./Session/agents/applied_session_agent";
+import { ServerWorker } from "./Session/agents/server_worker";
+import { Monitor } from "./Session/agents/monitor";
+import { MessageHandler } from "./Session/agents/promisified_ipc_manager";
/**
* If we're the monitor (master) thread, we should launch the monitor logic for the session.
diff --git a/src/server/DashSession/Session/agents/applied_session_agent.ts b/src/server/DashSession/Session/agents/applied_session_agent.ts
new file mode 100644
index 000000000..46c9e22ed
--- /dev/null
+++ b/src/server/DashSession/Session/agents/applied_session_agent.ts
@@ -0,0 +1,58 @@
+import { isMaster } from "cluster";
+import { Monitor } from "./monitor";
+import { ServerWorker } from "./server_worker";
+import { Utilities } from "../utilities/utilities";
+
+export type ExitHandler = (reason: Error | boolean) => void | Promise<void>;
+
+export abstract class AppliedSessionAgent {
+
+ // the following two methods allow the developer to create a custom
+ // session and use the built in customization options for each thread
+ protected abstract async initializeMonitor(monitor: Monitor): Promise<string>;
+ protected abstract async initializeServerWorker(): Promise<ServerWorker>;
+
+ private launched = false;
+
+ public killSession = (reason: string, graceful = true, errorCode = 0) => {
+ const target = isMaster ? this.sessionMonitor : this.serverWorker;
+ target.killSession(reason, graceful, errorCode);
+ }
+
+ private sessionMonitorRef: Monitor | undefined;
+ public get sessionMonitor(): Monitor {
+ if (!isMaster) {
+ this.serverWorker.emit("kill", {
+ graceful: false,
+ reason: "Cannot access the session monitor directly from the server worker thread.",
+ errorCode: 1
+ });
+ throw new Error();
+ }
+ return this.sessionMonitorRef!;
+ }
+
+ private serverWorkerRef: ServerWorker | undefined;
+ public get serverWorker(): ServerWorker {
+ if (isMaster) {
+ throw new Error("Cannot access the server worker directly from the session monitor thread");
+ }
+ return this.serverWorkerRef!;
+ }
+
+ public async launch(): Promise<void> {
+ if (!this.launched) {
+ this.launched = true;
+ if (isMaster) {
+ this.sessionMonitorRef = Monitor.Create()
+ const sessionKey = await this.initializeMonitor(this.sessionMonitorRef);
+ this.sessionMonitorRef.finalize(sessionKey);
+ } else {
+ this.serverWorkerRef = await this.initializeServerWorker();
+ }
+ } else {
+ throw new Error("Cannot launch a session thread more than once per process.");
+ }
+ }
+
+} \ No newline at end of file
diff --git a/src/server/DashSession/Session/agents/monitor.ts b/src/server/DashSession/Session/agents/monitor.ts
new file mode 100644
index 000000000..1d4ea6fb5
--- /dev/null
+++ b/src/server/DashSession/Session/agents/monitor.ts
@@ -0,0 +1,298 @@
+import { ExitHandler } from "./applied_session_agent";
+import { Configuration, configurationSchema, defaultConfig, Identifiers, colorMapping } from "../utilities/session_config";
+import Repl, { ReplAction } from "../utilities/repl";
+import { isWorker, setupMaster, on, Worker, fork } from "cluster";
+import { manage, MessageHandler } from "./promisified_ipc_manager";
+import { red, cyan, white, yellow, blue } from "colors";
+import { exec, ExecOptions } from "child_process";
+import { validate, ValidationError } from "jsonschema";
+import { Utilities } from "../utilities/utilities";
+import { readFileSync } from "fs";
+import IPCMessageReceiver from "./process_message_router";
+import { ServerWorker } from "./server_worker";
+
+/**
+ * Validates and reads the configuration file, accordingly builds a child process factory
+ * and spawns off an initial process that will respawn as predecessors die.
+ */
+export class Monitor extends IPCMessageReceiver {
+ private static count = 0;
+ private finalized = false;
+ private exitHandlers: ExitHandler[] = [];
+ private readonly config: Configuration;
+ private activeWorker: Worker | undefined;
+ private key: string | undefined;
+ private repl: Repl;
+
+ public static Create() {
+ if (isWorker) {
+ ServerWorker.IPCManager.emit("kill", {
+ reason: "cannot create a monitor on the worker process.",
+ graceful: false,
+ errorCode: 1
+ });
+ process.exit(1);
+ } else if (++Monitor.count > 1) {
+ console.error(red("cannot create more than one monitor."));
+ process.exit(1);
+ } else {
+ return new Monitor();
+ }
+ }
+
+ private constructor() {
+ super();
+ console.log(this.timestamp(), cyan("initializing session..."));
+ this.configureInternalHandlers();
+ this.config = this.loadAndValidateConfiguration();
+ this.initializeClusterFunctions();
+ this.repl = this.initializeRepl();
+ }
+
+ protected configureInternalHandlers = () => {
+ // handle exceptions in the master thread - there shouldn't be many of these
+ // the IPC (inter process communication) channel closed exception can't seem
+ // to be caught in a try catch, and is inconsequential, so it is ignored
+ process.on("uncaughtException", ({ message, stack }): void => {
+ if (message !== "Channel closed") {
+ this.mainLog(red(message));
+ if (stack) {
+ this.mainLog(`uncaught exception\n${red(stack)}`);
+ }
+ }
+ });
+
+ this.on("kill", ({ reason, graceful, errorCode }) => this.killSession(reason, graceful, errorCode));
+ this.on("lifecycle", ({ event }) => console.log(this.timestamp(), `${this.config.identifiers.worker.text} lifecycle phase (${event})`));
+ }
+
+ private initializeClusterFunctions = () => {
+ // determines whether or not we see the compilation / initialization / runtime output of each child server process
+ const output = this.config.showServerOutput ? "inherit" : "ignore";
+ setupMaster({ stdio: ["ignore", output, output, "ipc"] });
+
+ // a helpful cluster event called on the master thread each time a child process exits
+ on("exit", ({ process: { pid } }, code, signal) => {
+ const prompt = `server worker with process id ${pid} has exited with code ${code}${signal === null ? "" : `, having encountered signal ${signal}`}.`;
+ this.mainLog(cyan(prompt));
+ // to make this a robust, continuous session, every time a child process dies, we immediately spawn a new one
+ this.spawn();
+ });
+ }
+
+ public finalize = (sessionKey: string): void => {
+ if (this.finalized) {
+ throw new Error("Session monitor is already finalized");
+ }
+ this.finalized = true;
+ this.key = sessionKey;
+ this.spawn();
+ }
+
+ public readonly coreHooks = Object.freeze({
+ onCrashDetected: (listener: MessageHandler<{ error: Error }>) => this.on(Monitor.IntrinsicEvents.CrashDetected, listener),
+ onServerRunning: (listener: MessageHandler<{ isFirstTime: boolean }>) => this.on(Monitor.IntrinsicEvents.ServerRunning, listener)
+ });
+
+ /**
+ * Kill this session and its active child
+ * server process, either gracefully (may wait
+ * indefinitely, but at least allows active networking
+ * requests to complete) or immediately.
+ */
+ public killSession = async (reason: string, graceful = true, errorCode = 0) => {
+ this.mainLog(cyan(`exiting session ${graceful ? "clean" : "immediate"}ly`));
+ this.mainLog(`session exit reason: ${(red(reason))}`);
+ await this.executeExitHandlers(true);
+ await this.killActiveWorker(graceful, true);
+ process.exit(errorCode);
+ }
+
+ /**
+ * Execute the list of functions registered to be called
+ * whenever the process exits.
+ */
+ public addExitHandler = (handler: ExitHandler) => this.exitHandlers.push(handler);
+
+ /**
+ * Extend the default repl by adding in custom commands
+ * that can invoke application logic external to this module
+ */
+ public addReplCommand = (basename: string, argPatterns: (RegExp | string)[], action: ReplAction) => {
+ this.repl.registerCommand(basename, argPatterns, action);
+ }
+
+ public exec = (command: string, options?: ExecOptions) => {
+ return new Promise<void>(resolve => {
+ exec(command, { ...options, encoding: "utf8" }, (error, stdout, stderr) => {
+ if (error) {
+ this.execLog(red(`unable to execute ${white(command)}`));
+ error.message.split("\n").forEach(line => line.length && this.execLog(red(`(error) ${line}`)));
+ } else {
+ let outLines: string[], errorLines: string[];
+ if ((outLines = stdout.split("\n").filter(line => line.length)).length) {
+ outLines.forEach(line => line.length && this.execLog(cyan(`(stdout) ${line}`)));
+ }
+ if ((errorLines = stderr.split("\n").filter(line => line.length)).length) {
+ errorLines.forEach(line => line.length && this.execLog(yellow(`(stderr) ${line}`)));
+ }
+ }
+ resolve();
+ });
+ });
+ }
+
+ /**
+ * Generates a blue UTC string associated with the time
+ * of invocation.
+ */
+ private timestamp = () => blue(`[${new Date().toUTCString()}]`);
+
+ /**
+ * A formatted, identified and timestamped log in color
+ */
+ public mainLog = (...optionalParams: any[]) => {
+ console.log(this.timestamp(), this.config.identifiers.master.text, ...optionalParams);
+ }
+
+ /**
+ * A formatted, identified and timestamped log in color for non-
+ */
+ private execLog = (...optionalParams: any[]) => {
+ console.log(this.timestamp(), this.config.identifiers.exec.text, ...optionalParams);
+ }
+
+ /**
+ * Reads in configuration .json file only once, in the master thread
+ * and pass down any variables the pertinent to the child processes as environment variables.
+ */
+ private loadAndValidateConfiguration = (): Configuration => {
+ let config: Configuration;
+ try {
+ console.log(this.timestamp(), cyan("validating configuration..."));
+ config = JSON.parse(readFileSync('./session.config.json', 'utf8'));
+ const options = {
+ throwError: true,
+ allowUnknownAttributes: false
+ };
+ // ensure all necessary and no excess information is specified by the configuration file
+ validate(config, configurationSchema, options);
+ config = Utilities.preciseAssign({}, defaultConfig, config);
+ } catch (error) {
+ if (error instanceof ValidationError) {
+ console.log(red("\nSession configuration failed."));
+ console.log("The given session.config.json configuration file is invalid.");
+ console.log(`${error.instance}: ${error.stack}`);
+ process.exit(0);
+ } else if (error.code === "ENOENT" && error.path === "./session.config.json") {
+ console.log(cyan("Loading default session parameters..."));
+ console.log("Consider including a session.config.json configuration file in your project root for customization.");
+ config = Utilities.preciseAssign({}, defaultConfig);
+ } else {
+ console.log(red("\nSession configuration failed."));
+ console.log("The following unknown error occurred during configuration.");
+ console.log(error.stack);
+ process.exit(0);
+ }
+ } finally {
+ const { identifiers } = config!;
+ Object.keys(identifiers).forEach(key => {
+ const resolved = key as keyof Identifiers;
+ const { text, color } = identifiers[resolved];
+ identifiers[resolved].text = (colorMapping.get(color) || white)(`${text}:`);
+ });
+ return config!;
+ }
+ }
+
+ /**
+ * Builds the repl that allows the following commands to be typed into stdin of the master thread.
+ */
+ private initializeRepl = (): Repl => {
+ const repl = new Repl({ identifier: () => `${this.timestamp()} ${this.config.identifiers.master.text}` });
+ const boolean = /true|false/;
+ const number = /\d+/;
+ const letters = /[a-zA-Z]+/;
+ repl.registerCommand("exit", [/clean|force/], args => this.killSession("manual exit requested by repl", args[0] === "clean", 0));
+ repl.registerCommand("restart", [/clean|force/], args => this.killActiveWorker(args[0] === "clean"));
+ repl.registerCommand("set", [letters, "port", number, boolean], args => this.setPort(args[0], Number(args[2]), args[3] === "true"));
+ repl.registerCommand("set", [/polling/, number, boolean], args => {
+ const newPollingIntervalSeconds = Math.floor(Number(args[1]));
+ if (newPollingIntervalSeconds < 0) {
+ this.mainLog(red("the polling interval must be a non-negative integer"));
+ } else {
+ if (newPollingIntervalSeconds !== this.config.polling.intervalSeconds) {
+ this.config.polling.intervalSeconds = newPollingIntervalSeconds;
+ if (args[2] === "true") {
+ Monitor.IPCManager.emit("updatePollingInterval", { newPollingIntervalSeconds });
+ }
+ }
+ }
+ });
+ return repl;
+ }
+
+ private executeExitHandlers = async (reason: Error | boolean) => Promise.all(this.exitHandlers.map(handler => handler(reason)));
+
+ /**
+ * Attempts to kill the active worker gracefully, unless otherwise specified.
+ */
+ private killActiveWorker = async (graceful = true, isSessionEnd = false): Promise<void> => {
+ if (this.activeWorker && !this.activeWorker.isDead()) {
+ if (graceful) {
+ Monitor.IPCManager.emit("manualExit", { isSessionEnd });
+ } else {
+ await ServerWorker.IPCManager.destroy();
+ this.activeWorker.process.kill();
+ }
+ }
+ }
+
+ /**
+ * Allows the caller to set the port at which the target (be it the server,
+ * the websocket, some other custom port) is listening. If an immediate restart
+ * is specified, this monitor will kill the active child and re-launch the server
+ * at the port. Otherwise, the updated port won't be used until / unless the child
+ * dies on its own and triggers a restart.
+ */
+ private setPort = (port: "server" | "socket" | string, value: number, immediateRestart: boolean): void => {
+ if (value > 1023 && value < 65536) {
+ this.config.ports[port] = value;
+ if (immediateRestart) {
+ this.killActiveWorker();
+ }
+ } else {
+ this.mainLog(red(`${port} is an invalid port number`));
+ }
+ }
+
+ /**
+ * Kills the current active worker and proceeds to spawn a new worker,
+ * feeding in configuration information as environment variables.
+ */
+ private spawn = async (): Promise<void> => {
+ await this.killActiveWorker();
+ const { config: { polling, ports }, key } = this;
+ this.activeWorker = fork({
+ pollingRoute: polling.route,
+ pollingFailureTolerance: polling.failureTolerance,
+ serverPort: ports.server,
+ socketPort: ports.socket,
+ pollingIntervalSeconds: polling.intervalSeconds,
+ session_key: key
+ });
+ Monitor.IPCManager = manage(this.activeWorker.process, this.handlers);
+ this.mainLog(cyan(`spawned new server worker with process id ${this.activeWorker?.process.pid}`));
+ }
+
+}
+
+export namespace Monitor {
+
+ export enum IntrinsicEvents {
+ KeyGenerated = "key_generated",
+ CrashDetected = "crash_detected",
+ ServerRunning = "server_running"
+ }
+
+} \ No newline at end of file
diff --git a/src/server/DashSession/Session/agents/process_message_router.ts b/src/server/DashSession/Session/agents/process_message_router.ts
new file mode 100644
index 000000000..6cc8aa941
--- /dev/null
+++ b/src/server/DashSession/Session/agents/process_message_router.ts
@@ -0,0 +1,41 @@
+import { MessageHandler, PromisifiedIPCManager, HandlerMap } from "./promisified_ipc_manager";
+
+export default abstract class IPCMessageReceiver {
+
+ protected static IPCManager: PromisifiedIPCManager;
+ protected handlers: HandlerMap = {};
+
+ protected abstract configureInternalHandlers: () => void;
+
+ /**
+ * Add a listener at this message. When the monitor process
+ * receives a message, it will invoke all registered functions.
+ */
+ public on = (name: string, handler: MessageHandler) => {
+ const handlers = this.handlers[name];
+ if (!handlers) {
+ this.handlers[name] = [handler];
+ } else {
+ handlers.push(handler);
+ }
+ }
+
+ /**
+ * Unregister a given listener at this message.
+ */
+ public off = (name: string, handler: MessageHandler) => {
+ const handlers = this.handlers[name];
+ if (handlers) {
+ const index = handlers.indexOf(handler);
+ if (index > -1) {
+ handlers.splice(index, 1);
+ }
+ }
+ }
+
+ /**
+ * Unregister all listeners at this message.
+ */
+ public clearMessageListeners = (...names: string[]) => names.map(name => delete this.handlers[name]);
+
+} \ No newline at end of file
diff --git a/src/server/DashSession/Session/agents/promisified_ipc_manager.ts b/src/server/DashSession/Session/agents/promisified_ipc_manager.ts
new file mode 100644
index 000000000..9f0db8330
--- /dev/null
+++ b/src/server/DashSession/Session/agents/promisified_ipc_manager.ts
@@ -0,0 +1,173 @@
+import { Utilities } from "../utilities/utilities";
+import { ChildProcess } from "child_process";
+
+/**
+ * Convenience constructor
+ * @param target the process / worker to which to attach the specialized listeners
+ */
+export function manage(target: IPCTarget, handlers?: HandlerMap) {
+ return new PromisifiedIPCManager(target, handlers);
+}
+
+/**
+ * Captures the logic to execute upon receiving a message
+ * of a certain name.
+ */
+export type HandlerMap = { [name: string]: MessageHandler[] };
+
+/**
+ * This will always literally be a child process. But, though setting
+ * up a manager in the parent will indeed see the target as the ChildProcess,
+ * setting up a manager in the child will just see itself as a regular NodeJS.Process.
+ */
+export type IPCTarget = NodeJS.Process | ChildProcess;
+
+/**
+ * Specifies a general message format for this API
+ */
+export type Message<T = any> = {
+ name: string;
+ args?: T;
+};
+export type MessageHandler<T = any> = (args: T) => (any | Promise<any>);
+
+/**
+ * When a message is emitted, it is embedded with private metadata
+ * to facilitate the resolution of promises, etc.
+ */
+interface InternalMessage extends Message { metadata: Metadata }
+interface Metadata { isResponse: boolean; id: string }
+type InternalMessageHandler = (message: InternalMessage) => (any | Promise<any>);
+
+/**
+ * Allows for the transmission of the error's key features over IPC.
+ */
+export interface ErrorLike {
+ name?: string;
+ message?: string;
+ stack?: string;
+}
+
+/**
+ * The arguments returned in a message sent from the target upon completion.
+ */
+export interface Response<T = any> {
+ results?: T[];
+ error?: ErrorLike;
+}
+
+const destroyEvent = "__destroy__";
+
+/**
+ * This is a wrapper utility class that allows the caller process
+ * to emit an event and return a promise that resolves when it and all
+ * other processes listening to its emission of this event have completed.
+ */
+export class PromisifiedIPCManager {
+ private readonly target: IPCTarget;
+ private pendingMessages: { [id: string]: string } = {};
+ private isDestroyed = false;
+ private get callerIsTarget() {
+ return process.pid === this.target.pid;
+ }
+
+ constructor(target: IPCTarget, handlers?: HandlerMap) {
+ this.target = target;
+ if (handlers) {
+ handlers[destroyEvent] = [this.destroyHelper];
+ this.target.addListener("message", this.generateInternalHandler(handlers));
+ }
+ }
+
+ /**
+ * This routine uniquely identifies each message, then adds a general
+ * message listener that waits for a response with the same id before resolving
+ * the promise.
+ */
+ public emit = async <T = any>(name: string, args?: any): Promise<Response<T>> => {
+ if (this.isDestroyed) {
+ const error = { name: "FailedDispatch", message: "Cannot use a destroyed IPC manager to emit a message." };
+ return { error };
+ }
+ return new Promise<Response<T>>(resolve => {
+ const messageId = Utilities.guid();
+ const responseHandler: InternalMessageHandler = ({ metadata: { id, isResponse }, args }) => {
+ if (isResponse && id === messageId) {
+ this.target.removeListener("message", responseHandler);
+ resolve(args);
+ }
+ };
+ this.target.addListener("message", responseHandler);
+ const message = { name, args, metadata: { id: messageId, isResponse: false } };
+ if (!(this.target.send && this.target.send(message))) {
+ const error: ErrorLike = { name: "FailedDispatch", message: "Either the target's send method was undefined or the act of sending failed." };
+ resolve({ error });
+ this.target.removeListener("message", responseHandler);
+ }
+ });
+ }
+
+ /**
+ * Invoked from either the parent or the child process, this allows
+ * any unresolved promises to continue in the target process, but dispatches a dummy
+ * completion response for each of the pending messages, allowing their
+ * promises in the caller to resolve.
+ */
+ public destroy = () => {
+ return new Promise<void>(async resolve => {
+ if (this.callerIsTarget) {
+ this.destroyHelper();
+ } else {
+ await this.emit(destroyEvent);
+ }
+ resolve();
+ });
+ }
+
+ /**
+ * Dispatches the dummy responses and sets the isDestroyed flag to true.
+ */
+ private destroyHelper = () => {
+ const { pendingMessages } = this;
+ this.isDestroyed = true;
+ Object.keys(pendingMessages).forEach(id => {
+ const error: ErrorLike = { name: "ManagerDestroyed", message: "The IPC manager was destroyed before the response could be returned." };
+ const message: InternalMessage = { name: pendingMessages[id], args: { error }, metadata: { id, isResponse: true } };
+ this.target.send?.(message)
+ });
+ this.pendingMessages = {};
+ }
+
+ /**
+ * This routine receives a uniquely identified message. If the message is itself a response,
+ * it is ignored to avoid infinite mutual responses. Otherwise, the routine awaits its completion using whatever
+ * router the caller has installed, and then sends a response containing the original message id,
+ * which will ultimately invoke the responseHandler of the original emission and resolve the
+ * sender's promise.
+ */
+ private generateInternalHandler = (handlers: HandlerMap): MessageHandler => async (message: InternalMessage) => {
+ const { name, args, metadata } = message;
+ if (name && metadata && !metadata.isResponse) {
+ const { id } = metadata;
+ this.pendingMessages[id] = name;
+ let error: Error | undefined;
+ let results: any[] | undefined;
+ try {
+ const registered = handlers[name];
+ if (registered) {
+ results = await Promise.all(registered.map(handler => handler(args)));
+ }
+ } catch (e) {
+ error = e;
+ }
+ if (!this.isDestroyed && this.target.send) {
+ const metadata = { id, isResponse: true };
+ const response: Response = { results , error };
+ const message = { name, args: response , metadata };
+ delete this.pendingMessages[id];
+ this.target.send(message);
+ }
+ }
+ }
+
+} \ No newline at end of file
diff --git a/src/server/DashSession/Session/agents/server_worker.ts b/src/server/DashSession/Session/agents/server_worker.ts
new file mode 100644
index 000000000..976d27226
--- /dev/null
+++ b/src/server/DashSession/Session/agents/server_worker.ts
@@ -0,0 +1,160 @@
+import { ExitHandler } from "./applied_session_agent";
+import { isMaster } from "cluster";
+import { manage } from "./promisified_ipc_manager";
+import IPCMessageReceiver from "./process_message_router";
+import { red, green, white, yellow } from "colors";
+import { get } from "request-promise";
+import { Monitor } from "./monitor";
+
+/**
+ * Effectively, each worker repairs the connection to the server by reintroducing a consistent state
+ * if its predecessor has died. It itself also polls the server heartbeat, and exits with a notification
+ * email if the server encounters an uncaught exception or if the server cannot be reached.
+ */
+export class ServerWorker extends IPCMessageReceiver {
+ private static count = 0;
+ private shouldServerBeResponsive = false;
+ private exitHandlers: ExitHandler[] = [];
+ private pollingFailureCount = 0;
+ private pollingIntervalSeconds: number;
+ private pollingFailureTolerance: number;
+ private pollTarget: string;
+ private serverPort: number;
+ private isInitialized = false;
+
+ public static Create(work: Function) {
+ if (isMaster) {
+ console.error(red("cannot create a worker on the monitor process."));
+ process.exit(1);
+ } else if (++ServerWorker.count > 1) {
+ ServerWorker.IPCManager.emit("kill", {
+ reason: "cannot create more than one worker on a given worker process.",
+ graceful: false,
+ errorCode: 1
+ });
+ process.exit(1);
+ } else {
+ return new ServerWorker(work);
+ }
+ }
+
+ /**
+ * Allows developers to invoke application specific logic
+ * by hooking into the exiting of the server process.
+ */
+ public addExitHandler = (handler: ExitHandler) => this.exitHandlers.push(handler);
+
+ /**
+ * Kill the session monitor (parent process) from this
+ * server worker (child process). This will also kill
+ * this process (child process).
+ */
+ public killSession = (reason: string, graceful = true, errorCode = 0) => this.emit<never>("kill", { reason, graceful, errorCode });
+
+ /**
+ * A convenience wrapper to tell the session monitor (parent process)
+ * to carry out the action with the specified message and arguments.
+ */
+ public emit = async <T = any>(name: string, args?: any) => ServerWorker.IPCManager.emit<T>(name, args);
+
+ private constructor(work: Function) {
+ super();
+ this.configureInternalHandlers();
+ ServerWorker.IPCManager = manage(process, this.handlers);
+ this.lifecycleNotification(green(`initializing process... ${white(`[${process.execPath} ${process.execArgv.join(" ")}]`)}`));
+
+ const { pollingRoute, serverPort, pollingIntervalSeconds, pollingFailureTolerance } = process.env;
+ this.serverPort = Number(serverPort);
+ this.pollingIntervalSeconds = Number(pollingIntervalSeconds);
+ this.pollingFailureTolerance = Number(pollingFailureTolerance);
+ this.pollTarget = `http://localhost:${serverPort}${pollingRoute}`;
+
+ work();
+ this.pollServer();
+ }
+
+ /**
+ * Set up message and uncaught exception handlers for this
+ * server process.
+ */
+ protected configureInternalHandlers = () => {
+ // updates the local values of variables to the those sent from master
+ this.on("updatePollingInterval", ({ newPollingIntervalSeconds }) => this.pollingIntervalSeconds = newPollingIntervalSeconds);
+ this.on("manualExit", async ({ isSessionEnd }) => {
+ await ServerWorker.IPCManager.destroy();
+ await this.executeExitHandlers(isSessionEnd);
+ process.exit(0);
+ });
+
+ // one reason to exit, as the process might be in an inconsistent state after such an exception
+ process.on('uncaughtException', this.proactiveUnplannedExit);
+ process.on('unhandledRejection', reason => {
+ const appropriateError = reason instanceof Error ? reason : new Error(`unhandled rejection: ${reason}`);
+ this.proactiveUnplannedExit(appropriateError);
+ });
+ }
+
+ /**
+ * Execute the list of functions registered to be called
+ * whenever the process exits.
+ */
+ private executeExitHandlers = async (reason: Error | boolean) => Promise.all(this.exitHandlers.map(handler => handler(reason)));
+
+ /**
+ * Notify master thread (which will log update in the console) of initialization via IPC.
+ */
+ public lifecycleNotification = (event: string) => this.emit("lifecycle", { event });
+
+ /**
+ * Called whenever the process has a reason to terminate, either through an uncaught exception
+ * in the process (potentially inconsistent state) or the server cannot be reached.
+ */
+ private proactiveUnplannedExit = async (error: Error): Promise<void> => {
+ this.shouldServerBeResponsive = false;
+ // communicates via IPC to the master thread that it should dispatch a crash notification email
+ this.emit(Monitor.IntrinsicEvents.CrashDetected, { error });
+ await this.executeExitHandlers(error);
+ // notify master thread (which will log update in the console) of crash event via IPC
+ this.lifecycleNotification(red(`crash event detected @ ${new Date().toUTCString()}`));
+ this.lifecycleNotification(red(error.message));
+ await ServerWorker.IPCManager.destroy();
+ process.exit(1);
+ }
+
+ /**
+ * This monitors the health of the server by submitting a get request to whatever port / route specified
+ * by the configuration every n seconds, where n is also given by the configuration.
+ */
+ private pollServer = async (): Promise<void> => {
+ await new Promise<void>(resolve => {
+ setTimeout(async () => {
+ try {
+ await get(this.pollTarget);
+ if (!this.shouldServerBeResponsive) {
+ // notify monitor thread that the server is up and running
+ this.lifecycleNotification(green(`listening on ${this.serverPort}...`));
+ this.emit(Monitor.IntrinsicEvents.ServerRunning, { isFirstTime: !this.isInitialized });
+ this.isInitialized = true;
+ }
+ this.shouldServerBeResponsive = true;
+ } catch (error) {
+ // if we expect the server to be unavailable, i.e. during compilation,
+ // the listening variable is false, activeExit will return early and the child
+ // process will continue
+ if (this.shouldServerBeResponsive) {
+ if (++this.pollingFailureCount > this.pollingFailureTolerance) {
+ this.proactiveUnplannedExit(error);
+ } else {
+ this.lifecycleNotification(yellow(`the server has encountered ${this.pollingFailureCount} of ${this.pollingFailureTolerance} tolerable failures`));
+ }
+ }
+ } finally {
+ resolve();
+ }
+ }, 1000 * this.pollingIntervalSeconds);
+ });
+ // controlled, asynchronous infinite recursion achieves a persistent poll that does not submit a new request until the previous has completed
+ this.pollServer();
+ }
+
+} \ No newline at end of file
diff --git a/src/server/DashSession/Session/utilities/repl.ts b/src/server/DashSession/Session/utilities/repl.ts
new file mode 100644
index 000000000..643141286
--- /dev/null
+++ b/src/server/DashSession/Session/utilities/repl.ts
@@ -0,0 +1,128 @@
+import { createInterface, Interface } from "readline";
+import { red, green, white } from "colors";
+
+export interface Configuration {
+ identifier: () => string | string;
+ onInvalid?: (command: string, validCommand: boolean) => string | string;
+ onValid?: (success?: string) => string | string;
+ isCaseSensitive?: boolean;
+}
+
+export type ReplAction = (parsedArgs: Array<string>) => any | Promise<any>;
+export interface Registration {
+ argPatterns: RegExp[];
+ action: ReplAction;
+}
+
+export default class Repl {
+ private identifier: () => string | string;
+ private onInvalid: ((command: string, validCommand: boolean) => string) | string;
+ private onValid: ((success: string) => string) | string;
+ private isCaseSensitive: boolean;
+ private commandMap = new Map<string, Registration[]>();
+ public interface: Interface;
+ private busy = false;
+ private keys: string | undefined;
+
+ constructor({ identifier: prompt, onInvalid, onValid, isCaseSensitive }: Configuration) {
+ this.identifier = prompt;
+ this.onInvalid = onInvalid || this.usage;
+ this.onValid = onValid || this.success;
+ this.isCaseSensitive = isCaseSensitive ?? true;
+ this.interface = createInterface(process.stdin, process.stdout).on('line', this.considerInput);
+ }
+
+ private resolvedIdentifier = () => typeof this.identifier === "string" ? this.identifier : this.identifier();
+
+ private usage = (command: string, validCommand: boolean) => {
+ if (validCommand) {
+ const formatted = white(command);
+ const patterns = green(this.commandMap.get(command)!.map(({ argPatterns }) => `${formatted} ${argPatterns.join(" ")}`).join('\n'));
+ return `${this.resolvedIdentifier()}\nthe given arguments do not match any registered patterns for ${formatted}\nthe list of valid argument patterns is given by:\n${patterns}`;
+ } else {
+ const resolved = this.keys;
+ if (resolved) {
+ return resolved;
+ }
+ const members: string[] = [];
+ const keys = this.commandMap.keys();
+ let next: IteratorResult<string>;
+ while (!(next = keys.next()).done) {
+ members.push(next.value);
+ }
+ return `${this.resolvedIdentifier()} commands: { ${members.sort().join(", ")} }`;
+ }
+ }
+
+ private success = (command: string) => `${this.resolvedIdentifier()} completed local execution of ${white(command)}`;
+
+ public registerCommand = (basename: string, argPatterns: (RegExp | string)[], action: ReplAction) => {
+ const existing = this.commandMap.get(basename);
+ const converted = argPatterns.map(input => input instanceof RegExp ? input : new RegExp(input));
+ const registration = { argPatterns: converted, action };
+ if (existing) {
+ existing.push(registration);
+ } else {
+ this.commandMap.set(basename, [registration]);
+ }
+ }
+
+ private invalid = (command: string, validCommand: boolean) => {
+ console.log(red(typeof this.onInvalid === "string" ? this.onInvalid : this.onInvalid(command, validCommand)));
+ this.busy = false;
+ }
+
+ private valid = (command: string) => {
+ console.log(green(typeof this.onValid === "string" ? this.onValid : this.onValid(command)));
+ this.busy = false;
+ }
+
+ private considerInput = async (line: string) => {
+ if (this.busy) {
+ console.log(red("Busy"));
+ return;
+ }
+ this.busy = true;
+ line = line.trim();
+ if (this.isCaseSensitive) {
+ line = line.toLowerCase();
+ }
+ const [command, ...args] = line.split(/\s+/g);
+ if (!command) {
+ return this.invalid(command, false);
+ }
+ const registered = this.commandMap.get(command);
+ if (registered) {
+ const { length } = args;
+ const candidates = registered.filter(({ argPatterns: { length: count } }) => count === length);
+ for (const { argPatterns, action } of candidates) {
+ const parsed: string[] = [];
+ let matched = true;
+ if (length) {
+ for (let i = 0; i < length; i++) {
+ let matches: RegExpExecArray | null;
+ if ((matches = argPatterns[i].exec(args[i])) === null) {
+ matched = false;
+ break;
+ }
+ parsed.push(matches[0]);
+ }
+ }
+ if (!length || matched) {
+ const result = action(parsed);
+ const resolve = () => this.valid(`${command} ${parsed.join(" ")}`);
+ if (result instanceof Promise) {
+ result.then(resolve);
+ } else {
+ resolve();
+ }
+ return;
+ }
+ }
+ this.invalid(command, true);
+ } else {
+ this.invalid(command, false);
+ }
+ }
+
+} \ No newline at end of file
diff --git a/src/server/DashSession/Session/utilities/session_config.ts b/src/server/DashSession/Session/utilities/session_config.ts
new file mode 100644
index 000000000..b0e65dde4
--- /dev/null
+++ b/src/server/DashSession/Session/utilities/session_config.ts
@@ -0,0 +1,129 @@
+import { Schema } from "jsonschema";
+import { yellow, red, cyan, green, blue, magenta, Color, grey, gray, white, black } from "colors";
+
+const colorPattern = /black|red|green|yellow|blue|magenta|cyan|white|gray|grey/;
+
+const identifierProperties: Schema = {
+ type: "object",
+ properties: {
+ text: {
+ type: "string",
+ minLength: 1
+ },
+ color: {
+ type: "string",
+ pattern: colorPattern
+ }
+ }
+};
+
+const portProperties: Schema = {
+ type: "number",
+ minimum: 1024,
+ maximum: 65535
+};
+
+export const configurationSchema: Schema = {
+ id: "/configuration",
+ type: "object",
+ properties: {
+ showServerOutput: { type: "boolean" },
+ ports: {
+ type: "object",
+ properties: {
+ server: portProperties,
+ socket: portProperties
+ },
+ required: ["server"],
+ additionalProperties: true
+ },
+ identifiers: {
+ type: "object",
+ properties: {
+ master: identifierProperties,
+ worker: identifierProperties,
+ exec: identifierProperties
+ }
+ },
+ polling: {
+ type: "object",
+ additionalProperties: false,
+ properties: {
+ intervalSeconds: {
+ type: "number",
+ minimum: 1,
+ maximum: 86400
+ },
+ route: {
+ type: "string",
+ pattern: /\/[a-zA-Z]*/g
+ },
+ failureTolerance: {
+ type: "number",
+ minimum: 0,
+ }
+ }
+ },
+ }
+};
+
+type ColorLabel = "yellow" | "red" | "cyan" | "green" | "blue" | "magenta" | "grey" | "gray" | "white" | "black";
+
+export const colorMapping: Map<ColorLabel, Color> = new Map([
+ ["yellow", yellow],
+ ["red", red],
+ ["cyan", cyan],
+ ["green", green],
+ ["blue", blue],
+ ["magenta", magenta],
+ ["grey", grey],
+ ["gray", gray],
+ ["white", white],
+ ["black", black]
+]);
+
+interface Identifier {
+ text: string;
+ color: ColorLabel;
+}
+
+export interface Identifiers {
+ master: Identifier;
+ worker: Identifier;
+ exec: Identifier;
+}
+
+export interface Configuration {
+ showServerOutput: boolean;
+ identifiers: Identifiers;
+ ports: { [description: string]: number };
+ polling: {
+ route: string;
+ intervalSeconds: number;
+ failureTolerance: number;
+ };
+}
+
+export const defaultConfig: Configuration = {
+ showServerOutput: false,
+ identifiers: {
+ master: {
+ text: "__monitor__",
+ color: "yellow"
+ },
+ worker: {
+ text: "__server__",
+ color: "magenta"
+ },
+ exec: {
+ text: "__exec__",
+ color: "green"
+ }
+ },
+ ports: { server: 3000 },
+ polling: {
+ route: "/",
+ intervalSeconds: 30,
+ failureTolerance: 0
+ }
+}; \ No newline at end of file
diff --git a/src/server/DashSession/Session/utilities/utilities.ts b/src/server/DashSession/Session/utilities/utilities.ts
new file mode 100644
index 000000000..eb8de9d7e
--- /dev/null
+++ b/src/server/DashSession/Session/utilities/utilities.ts
@@ -0,0 +1,37 @@
+import { v4 } from "uuid";
+
+export namespace Utilities {
+
+ export function guid() {
+ return v4();
+ }
+
+ /**
+ * At any arbitrary layer of nesting within the configuration objects, any single value that
+ * is not specified by the configuration is given the default counterpart. If, within an object,
+ * one peer is given by configuration and two are not, the one is preserved while the two are given
+ * the default value.
+ * @returns the composition of all of the assigned objects, much like Object.assign(), but with more
+ * granularity in the overwriting of nested objects
+ */
+ export function preciseAssign(target: any, ...sources: any[]): any {
+ for (const source of sources) {
+ preciseAssignHelper(target, source);
+ }
+ return target;
+ }
+
+ export function preciseAssignHelper(target: any, source: any) {
+ Array.from(new Set([...Object.keys(target), ...Object.keys(source)])).map(property => {
+ let targetValue: any, sourceValue: any;
+ if (sourceValue = source[property]) {
+ if (typeof sourceValue === "object" && typeof (targetValue = target[property]) === "object") {
+ preciseAssignHelper(targetValue, sourceValue);
+ } else {
+ target[property] = sourceValue;
+ }
+ }
+ });
+ }
+
+} \ No newline at end of file
diff --git a/src/server/index.ts b/src/server/index.ts
index 2101de1d2..88f5fa3bf 100644
--- a/src/server/index.ts
+++ b/src/server/index.ts
@@ -24,7 +24,7 @@ import { Logger } from "./ProcessFactory";
import { yellow } from "colors";
import { DashSessionAgent } from "./DashSession/DashSessionAgent";
import SessionManager from "./ApiManagers/SessionManager";
-import { AppliedSessionAgent } from "resilient-server-session";
+import { AppliedSessionAgent } from "./DashSession/Session/agents/applied_session_agent";
export const onWindows = process.platform === "win32";
export let sessionAgent: AppliedSessionAgent;
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2025-10-06 13:10:30 +0000