Replit Commit

lib/sockets/clients/Executor.js

@@ -0,0 +1,75 @@
+import { Manager } from "socket.io-client";
+import cp from "child_process";
+
+import modes from "../modes.js";
+import events from "../events.js";
+
+export { default as events } from "../events.js";
+export { default as modes } from "../modes.js";
+
+// Data Stream Types
+const ERR = "e";
+const OUT = "o";
+
+export default class Executor {
+  constructor(url, job, options = {}) {
+    this.url = url;
+    this.job = job;
+    this.mode = modes.EXEC;
+
+    // Internal Buffer
+    this.buf = {};
+    this.buf[ERR] = "";
+    this.buf[OUT] = "";
+
+    // Methods
+    this.spawn = options.spawn ?? this.spawn.bind(this);
+    this.report = options.report ?? this.report.bind(this);
+    this.onProcClose = options.onProcClose ?? this.onProcClose.bind(this);
+    this.onClose = options.onClose ?? this.onClose.bind(this);
+  }
+
+  spawn() {
+    const cmdArgs = this.job.command.split(" ");
+    const cmd = cmdArgs.shift();
+    this.proc = cp.spawn(cmd, cmdArgs);
+
+    // Set Encoding
+    this.proc.stdout.setEncoding("utf8");
+    this.proc.stderr.setEncoding("utf8");
+
+    // Process Events
+    this.proc.stdout.on("data", (d) => this.report(d.toString(), OUT));
+    this.proc.stderr.on("data", (d) => this.report(d.toString(), ERR));
+    this.proc.on("close", this.onProcClose);
+  }
+
+  runJob() {
+    const mgr = new Manager(this.url, {
+      query: { mode: this.mode, jobId: this.job.id },
+    });
+    this.socket = mgr.socket("/");
+    this.socket.on("connect", this.spawn);
+    this.socket.on("disconnect", this.onClose);
+  }
+
+  onClose() {
+    console.log("Server disconnected, terminating process.");
+    this.proc.kill("SIGINT");
+  }
+
+  onProcClose(code) {
+    this.socket.emit(events.JOB_CLS, code);
+    console.log(`Process finished with code ${code}`);
+    this.socket.disconnect();
+  }
+
+  report(d, dType) {
+    this.buf[dType] += d;
+    if (!this.buf[dType].includes("\n")) return;
+    this.socket.emit(events.JOB_REP, this.buf[dType]);
+    if (dType === ERR) console.error(`err: ${this.buf[dType]}`);
+    else console.log(`out: ${this.buf[dType]}`);
+    this.buf[dType] = "";
+  }
+}

lib/sockets/clients/Initiator.js

@@ -0,0 +1,34 @@
+import { Manager } from "socket.io-client";
+import modes from "../modes.js";
+import events from "../events.js";
+
+export { default as events } from "../events.js";
+export { default as modes } from "../modes.js";
+
+export default class Initiator {
+  constructor(url, options = {}) {
+    this.url = url;
+    this.mode = modes.INIT;
+    this.onLog = options.onLog ?? ((d) => console.log(`job: ${d}`));
+    this.onClose = options.onClose ?? (() => {});
+    this.onCreate = options.onCreate ?? ((id) => console.log(`job id: ${id}`));
+  }
+
+  async newJob(jobRequest, onLog, onClose, onCreate) {
+    const mgr = new Manager(this.url, {
+      query: { mode: this.mode, job: JSON.stringify(jobRequest) },
+    });
+    onLog = onLog ?? this.onLog.bind(this);
+    onClose = onClose ?? this.onClose.bind(this);
+    onCreate = onCreate ?? this.onCreate.bind(this);
+    const sk = mgr.socket("/");
+    sk.on(events.JOB_LOG, onLog);
+    sk.on(events.JOB_CLS, onClose);
+    return new Promise((res) =>
+      sk.on(events.JOB_CRT, function onJobCreate(id) {
+        onCreate(id);
+        res({ ...jobRequest, id });
+      })
+    );
+  }
+}

lib/sockets/clients/Viewer.js

@@ -0,0 +1,27 @@
+import { Manager } from "socket.io-client";
+import modes from "../modes.js";
+import events from "../events.js";
+
+export { default as events } from "../events.js";
+export { default as modes } from "../modes.js";
+
+export default class Viewer {
+  constructor(url, options = {}) {
+    this.url = url;
+    this.mode = modes.VIEW;
+    this.onLog = options.onLog ?? console.log;
+    this.onClose = options.onClose ?? (() => {});
+  }
+
+  viewJob(jobId, onLog, onClose) {
+    const mgr = new Manager(this.url, {
+      query: { mode: this.mode, jobId },
+    });
+    onLog = onLog ?? this.onLog.bind(this);
+    onClose = onClose ?? this.onClose.bind(this);
+    const sk = mgr.socket("/");
+    sk.on(events.JOB_LOG, onLog);
+    sk.on(events.JOB_CLS, onClose);
+    return sk;
+  }
+}

lib/sockets/clients/index.js

@@ -0,0 +1,5 @@
+export { default as Initiator } from "./Initiator.js";
+
+export { default as Viewer } from "./Viewer.js";
+
+export { default as Executor } from "./Executor.js";

lib/sockets/clients/web.index.js

@@ -0,0 +1,3 @@
+export { default as Initiator } from "./Initiator.js";
+
+export { default as Viewer } from "./Viewer.js";

lib/sockets/events.js

@@ -0,0 +1,13 @@
+const JOB_REP = "jr"; // Job Report Event
+const JOB_LOG = "jl"; // Job Log Event
+const JOB_CLS = "jc"; // Job Close Event
+const JOB_CRT = "jcr"; // Job Create Event
+const ERR = "e"; // Socket Error
+
+export default {
+  JOB_REP,
+  JOB_LOG,
+  JOB_CLS,
+  JOB_CRT,
+  ERR,
+};

lib/sockets/handler.js

@@ -0,0 +1,48 @@
+import { Server as Skio } from "socket.io";
+import evt from "./events.js";
+import modes from "./modes.js";
+
+import { initiator, executor, viewer } from "./modifiers.js";
+
+const socketDrop = (io, room, id) => {
+  const { rooms } = io.of("/").adapter;
+  const clients = rooms.get(room);
+  if (clients.size > 1 || clients.size === 0) return;
+  const socketId = Array.from(clients)[0];
+  const s = io.sockets.sockets.get(socketId);
+  s.disconnect();
+};
+
+const socketConnect = (io, socket, jobs) => {
+  const { mode } = socket.handshake.query;
+  switch (mode) {
+    case modes.INIT:
+      initiator(socket, jobs);
+      break;
+    case modes.EXEC:
+      executor(io, socket, jobs);
+      break;
+    case modes.VIEW:
+      viewer(socket);
+      break;
+    default:
+      socket.send(evt.ERR, "Invalid Mode!");
+      socket.disconnect();
+      break;
+  }
+};
+
+const socketAuth = (socket, next) => {
+  const { token } = socket.handshake.auth;
+  // next(new Error("Bad Token"));
+  next();
+};
+
+const applySockets = (server, jobs, options) => {
+  const io = new Skio(server);
+  io.on("connection", (socket) => socketConnect(io, socket, jobs));
+  io.of("/").adapter.on("leave-room", (room, id) => socketDrop(io, room, id));
+  return io;
+};
+
+export default applySockets;

lib/sockets/modes.js

@@ -0,0 +1,8 @@
+const INIT = "i"; // Intiator Socket
+const EXEC = "e"; // Execution Socket
+const VIEW = "v"; // View Socket
+export default {
+  INIT,
+  EXEC,
+  VIEW,
+};

lib/sockets/modifiers.js

@@ -0,0 +1,34 @@
+import evt from "./events.js";
+
+export const initiator = (socket, jobs) => {
+  const jobStr = socket.handshake.query.job;
+  const jobReq = JSON.parse(jobStr);
+
+  if (!jobReq || !(jobReq instanceof Object))
+    throw Error("No 'job' was included in the handshake query");
+
+  const job = jobs.newJob(jobReq, socket.id);
+  socket.join(job.id);
+  socket.emit(evt.JOB_CRT, job.id);
+};
+
+export const executor = (io, socket, jobs) => {
+  const jobId = socket.handshake.query.jobId;
+  if (!jobId) throw Error("No 'jobId' was included in the handshake query");
+
+  socket.join(jobId);
+  socket.on(evt.JOB_REP, function onReport(log) {
+    jobs.pushLog(jobId, log);
+    io.to(jobId).emit(evt.JOB_LOG, log);
+  });
+  socket.on(evt.JOB_CLS, function onClose(code) {
+    jobs.closeJob(jobId, code);
+    io.to(jobId).emit(evt.JOB_CLS, code);
+  });
+};
+
+export const viewer = (socket) => {
+  const jobId = socket.handshake.query.jobId;
+  if (!jobId) throw Error("No 'jobId' was included in the handshake query");
+  socket.join(jobId);
+};