Switched to an image API implementation by Terradice

api/index.js

@@ -1,23 +1,95 @@
-require("dotenv").config();
+// code provided by terradice/tzlil
+
+const os = require("os");
+const { Worker, isMainThread, parentPort } = require("worker_threads");
 const magick = require("../utils/image.js");
-const Job = require("./job.js");
-const { version } = require("../package.json");
-const express = require("express");
 const execPromise = require("util").promisify(require("child_process").exec);
-const app = express();
+const net = require("net");
-const port = 3000;
+const dgram = require("dgram"); // for UDP servers
+const socket = dgram.createSocket("udp4"); // Our universal UDP socket, this might cause issues and we may have to use a seperate socket for each connection
 
-const jobs = new Map();
+var start = process.hrtime();
+const log = (msg, thread) => {
+  console.log(`[${process.hrtime(start)[1] / 1000000}${(thread)?`:${thread}`:""}]\t ${msg}`);
+};
 
-app.get("/", (req, res) => {
+const jobs = {};
-  res.send(`esmBot v${version}`);
+// Should look like UUID : { addr : "someaddr", port: someport msg: "request" }
-});
+const queue = [];
+// Array of UUIDs
 
-app.post("/run", express.json(), async (req, res, next) => {
+if (isMainThread) {
-  const object = req.body;
+  const { v4: uuidv4 } = require("uuid");
-  if (!magick.check(object.cmd)) return res.sendStatus(400);
+
+  const MAX_WORKERS = process.env.WORKERS === "" ? parseInt(process.env.WORKERS) : os.cpus().length * 4; // Completely arbitrary, should usually be some multiple of your amount of cores
+  let workingWorkers = 0;
+
+  const acceptJob = (uuid) => {
+    workingWorkers++;
+    const worker = new Worker(__filename);
+    queue.shift();
+    worker.once("message", (uuid) => {
+      // This means the worker is finished
+      workingWorkers--;
+      if (queue.length > 0) {
+        acceptJob(queue[0]); // Get the next job UUID in queue and remove it from the original queue
+        delete jobs[uuid];
+      }
+    });
+    worker.on("error", err => console.error("worker error:", err));
+    worker.on("exit", (code) => {
+      if (code !== 0)
+        console.error(`Worker stopped with exit code ${code}`);
+    });
+
+
+    worker.postMessage({
+      uuid: uuid,
+      msg: jobs[uuid].msg,
+      addr: jobs[uuid].addr,
+      port: jobs[uuid].port,
+      threadNum: workingWorkers
+    });
+  };
+
+  const server = dgram.createSocket("udp4"); //Create a UDP server for listening to requests, we dont need tcp
+  server.on("message", (msg, rinfo) => {
+    const opcode = msg.readUint8(0);
+    const req = msg.toString().slice(1,msg.length);
+    // 0x0 == Cancel job
+    // 0x1 == Queue job
+    if (opcode == 0x0) {
+      queue.shift();
+      delete jobs[req];
+    } else if (opcode == 0x1) {
+      const job = { addr: rinfo.address, port: rinfo.port, msg: req };
+      const uuid = uuidv4();
+      jobs[uuid] = job;
+      queue.push(uuid);
+
+      if (workingWorkers < MAX_WORKERS) {
+        acceptJob(uuid);
+      }
+
+      const newBuffer = Buffer.concat([Buffer.from([0x0]), Buffer.from(uuid)]);
+      socket.send(newBuffer, rinfo.port, rinfo.address);
+    } else {
+      log("Could not parse message");
+    }
+  });
+
+  server.on("listening", () => {
+    const address = server.address();
+    log(`server listening ${address.address}:${address.port}`);
+  });
+
+  server.bind(8080); // ATTENTION: Always going to be bound to 0.0.0.0 !!!
+} else {
+  parentPort.once("message", async (job) => {
+    log(`${job.uuid} worker got: ${job.msg}`, job.threadNum);
 
     try {
+      const object = JSON.parse(job.msg);
       let type;
       if (object.path) {
         type = object.type;
@@ -25,74 +97,40 @@ app.post("/run", express.json(), async (req, res, next) => {
           type = await magick.getType(object.path);
         }
         if (!type) {
-        return res.sendStatus(400);
+          throw new TypeError("Unknown image type");
         }
         object.type = type.split("/")[1];
-      if (object.type !== "gif" && object.onlyGIF) return res.send({
+        if (object.type !== "gif" && object.onlyGIF) throw new TypeError(`Expected a GIF, got ${object.type}`);
-        status: "nogif"
-      });
         object.delay = object.delay ? object.delay : 0;
       }
       
-    const id = Math.random().toString(36).substring(2, 15);
       if (object.type === "gif" && !object.delay) {
         const delay = (await execPromise(`ffprobe -v 0 -of csv=p=0 -select_streams v:0 -show_entries stream=r_frame_rate ${object.path}`)).stdout.replace("\n", "");
         object.delay = (100 / delay.split("/")[0]) * delay.split("/")[1];
       }
-    const job = new Job(object);
+
-    jobs.set(id, job);
+      const data = await magick.run(object, true);
-    res.send({
+
-      id: id,
+      log(`${job.uuid} is done`, job.threadNum);
-      status: "queued"
+      const server = net.createServer(function(socket) {
+        socket.write(Buffer.concat([Buffer.from(type ? type : "image/png"), Buffer.from("\n"), data]));
       });
-    job.run();
+      server.listen(job.port, job.addr);
-  } catch (e) {
+      // handle address in use errors
-    next(e);
+      server.on("error", (e) => {
+        if (e.code === "EADDRINUSE") {
+          console.log("Address in use, retrying...");
+          setTimeout(() => {
+            server.close();
+            server.listen(job.port, job.addr);
+          }, 500);
         }
-});
-
-app.get("/status", (req, res) => {
-  if (!req.query.id) return res.sendStatus(400);
-  const job = jobs.get(req.query.id);
-  if (!job) return res.sendStatus(400);
-  const timeout = setTimeout(function() {
-    job.removeAllListeners();
-    return res.send({
-      id: req.query.id,
-      status: job.status
       });
-  }, 10000);
+      socket.send(Buffer.concat([Buffer.from([0x1]), Buffer.from(job.port.toString())]), job.port, job.addr);
-  job.once("data", function() {
+      parentPort.postMessage(job.uuid); //Inform main thread about this worker freeing up
-    clearTimeout(timeout);
+    } catch (e) {
-    res.send({
+      socket.send(Buffer.concat([Buffer.from([0x2]), Buffer.from(e.toString())]), job.port, job.address);
-      id: req.query.id,
+      parentPort.postMessage(job.uuid);
-      status: job.status
+    }
   });
-    //jobs.delete(req.query.id);
+}
-  });
-  job.on("error", function(e) {
-    clearTimeout(timeout);
-    res.status(500);
-    res.send({
-      id: req.query.id,
-      status: job.status,
-      error: e
-    });
-    jobs.delete(req.query.id);
-  });
-});
-
-app.get("/image", (req, res) => {
-  if (!req.query.id) return res.sendStatus(400);
-  const job = jobs.get(req.query.id);
-  if (!job) return res.sendStatus(400);
-  if (!job.data) return res.sendStatus(400);
-  if (job.error) return;
-  jobs.delete(req.query.id);
-  res.contentType(job.options.type ? job.options.type : "png");
-  return res.send(job.data);
-});
-
-app.listen(port, () => {
-  console.log(`Started image API on port ${port}.`);
-});

api/job.js

@@ -1,28 +0,0 @@
-const { EventEmitter } = require("events");
-const magick = require("../utils/image.js");
-
-class Job extends EventEmitter {
-  constructor(options) {
-    super();
-    this.options = options;
-    this.status = "queued";
-    this.data = null;
-    this.error = null;
-  }
-
-  run() {
-    this.status = "processing";
-    magick.run(this.options, true).then(data => {
-      this.status = "success";
-      this.data = data;
-      return this.emit("data", data, this.options.type);
-    }).catch(e => {
-      this.status = "error";
-      this.error = e;
-      return this.emit("error", e);
-    });
-    return;
-  }
-}
-
-module.exports = Job;

package-lock.json

@@ -2904,6 +2904,12 @@
       "integrity": "sha1-n5VxD1CiZ5R7LMwSR0HBAoQn5xM=",
       "optional": true
     },
+    "uuid": {
+      "version": "8.3.1",
+      "resolved": "https://registry.npmjs.org/uuid/-/uuid-8.3.1.tgz",
+      "integrity": "sha512-FOmRr+FmWEIG8uhZv6C2bTgEVXsHk08kE7mPlrBbEe+c3r9pjceVPgupIfNIhc4yx55H69OXANrUaSuu9eInKg==",
+      "optional": true
+    },
     "vary": {
       "version": "1.1.2",
       "resolved": "https://registry.npmjs.org/vary/-/vary-1.1.2.tgz",
@@ -3036,9 +3042,9 @@
       }
     },
     "ws": {
-      "version": "7.3.1",
+      "version": "7.4.0",
-      "resolved": "https://registry.npmjs.org/ws/-/ws-7.3.1.tgz",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-7.4.0.tgz",
-      "integrity": "sha512-D3RuNkynyHmEJIpD2qrgVkc9DQ23OrN/moAwZX4L8DfvszsJxpjQuUq3LMx6HoYji9fbIOBY18XWBsAux1ZZUA=="
+      "integrity": "sha512-kyFwXuV/5ymf+IXhS6f0+eAFvydbaBW3zjpT6hUdAh/hbVjTIB5EHBGi0bPoCLSK2wcuz3BrEkB9LrYv1Nm4NQ=="
     },
     "y18n": {
       "version": "4.0.0",

package.json

@@ -54,6 +54,7 @@
     "bufferutil": "^4.0.1",
     "erlpack": "github:abalabahaha/erlpack",
     "express": "^4.17.1",
+    "uuid": "^8.3.1",
     "zlib-sync": "^0.1.6"
   }
 }

servers.json

@@ -3,6 +3,6 @@
     { "id": "1", "host": "localhost", "port": 2333, "password": "youshallnotpass" }
   ],
   "image": [
-    "http://localhost:3000"
+    "localhost"
   ]
 }

utils/image.js

@@ -2,61 +2,88 @@ const magick = require("../build/Release/image.node");
 const fetch = require("node-fetch");
 const { promisify } = require("util");
 const AbortController = require("abort-controller");
+const net = require("net");
+const dgram = require("dgram");
 const fileType = require("file-type");
 const execPromise = promisify(require("child_process").exec);
 const servers = require("../servers.json").image;
 
 const formats = ["image/jpeg", "image/png", "image/webp", "image/gif"];
 
-exports.run = async (object, fromAPI = false) => {
+const getFormat = (buffer, delimiter) => {
+  for (var i = 0; i < buffer.length ; i++) {
+    if (String.fromCharCode(buffer[i]) === delimiter) {
+      return {
+        buffer: buffer.slice(0, i),
+        dataStart: i
+      };
+    }
+  }
+};
+
+exports.run = (object, fromAPI = false) => {
+  return new Promise(async (resolve, reject) => {
     if (process.env.API === "true" && !fromAPI) {
       const currentServer = servers[Math.floor(Math.random() * servers.length)];
-    const req = await fetch(`${currentServer}/run`, {
+      const socket = dgram.createSocket("udp4");
-      method: "POST",
+      const data = Buffer.concat([Buffer.from([0x1]), Buffer.from(JSON.stringify(object))]);
-      body: JSON.stringify(object),
+  
-      headers: {
+      //let jobID;
-        "Content-Type": "application/json"
+      socket.on("message", (msg) => {
+        const opcode = msg.readUint8(0);
+        const req = msg.slice(1, msg.length);
+        if (opcode === 0x0) {
+          //jobID = req;
+          //console.log(`Our job UUID is: ${jobID}`);
+        } else if (opcode === 0x1) {
+          //console.log(`Job ${jobID} is finished!`);
+          const client = net.createConnection(req.toString(), currentServer);
+          const array = [];
+          client.on("data", (rawData) => {
+            array.push(rawData);
+            if (rawData.length !== 32 * 1024) {
+              client.end();
             }
           });
-    const json = await req.json();
+          client.once("end", () => {
-    if (json.status === "nogif") return {
+            const data = Buffer.concat(array);
-      buffer: "nogif",
+            const format = getFormat(data, "\n");
-      type: null
+            const payload = {
+              buffer: data.slice(format.dataStart + 1),
+              type: format.buffer.toString().split("/")[1]
             };
-    let data;
+            //console.log(payload);
-    while (!data) {
+            socket.close();
-      const statusReq = await fetch(`${currentServer}/status?id=${json.id}`);
+            resolve(payload);
-      const statusJSON = await statusReq.json();
+          });
-      if (statusJSON.status === "success") {
+        } else if (opcode === 0x2) {
-        const imageReq = await fetch(`${currentServer}/image?id=${json.id}`);
+          reject(req);
-        data = {
-          buffer: await imageReq.buffer(),
-          type: imageReq.headers.get("content-type").split("/")[1]
-        };
-      } else if (statusJSON.status === "error") {
-        throw new Error(statusJSON.error);
         }
-    }
+      });
-    return data;
+  
+      socket.send(data, 8080, currentServer, (err) => {
+        if (err) reject(err);
+      });
     } else {
       let type;
       if (!fromAPI && object.path) {
         const newType = (object.type ? object.type : await this.getType(object.path));
         type = newType ? newType.split("/")[1] : "png";
-      if (type !== "gif" && object.onlyGIF) return {
+        if (type !== "gif" && object.onlyGIF) resolve({
           buffer: "nogif",
           type: null
-      };
+        });
         object.type = type;
         const delay = (await execPromise(`ffprobe -v 0 -of csv=p=0 -select_streams v:0 -show_entries stream=r_frame_rate ${object.path}`)).stdout.replace("\n", "");
         object.delay = (100 / delay.split("/")[0]) * delay.split("/")[1];
       }
       const data = await promisify(magick[object.cmd])(object);
-    return fromAPI ? data : {
+      resolve(fromAPI ? data : {
         buffer: data,
         type: type
-    };
+      });
     }
+  });
 };
 
 exports.check = (cmd) => {