murm/mrmBot-Matrix
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Handle errors properly *and* also hopefully fix the file descriptor issue for real

Browse source
This commit is contained in:
TheEssem 2021-01-02 23:56:27 -06:00
parent ae32a3f9f1
commit be53f8d352
1 changed files with 40 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

91
api/index.js
View file

@ -64,14 +64,8 @@ if (isMainThread) {
} }
}); });
worker.once("error", err => { worker.once("error", err => {
console.error("worker error:", err); console.error(`Error on worker ${uuid}:`, err);
socket.send(Buffer.concat([Buffer.from([0x2]), Buffer.from(uuid), Buffer.from(err.toString())]), jobs[uuid].port, jobs[uuid].addr); socket.send(Buffer.concat([Buffer.from([0x2]), Buffer.from(uuid), Buffer.from(err.toString())]), jobs[uuid].port, jobs[uuid].addr);
workingWorkers--;
if (queue.length > 0) {
acceptJob(queue[0]);
delete jobs[uuid];
}
}); });
worker.once("exit", (code) => { worker.once("exit", (code) => {
workingWorkers--; workingWorkers--;
@ -80,7 +74,7 @@ if (isMainThread) {
delete jobs[uuid]; delete jobs[uuid];
} }
if (code !== 0) console.error(`Worker stopped with exit code ${code}`); if (code !== 0) console.error(`Worker ${uuid} stopped with exit code ${code}`);
}); });
@ -135,54 +129,49 @@ if (isMainThread) {
parentPort.once("message", async (job) => { parentPort.once("message", async (job) => {
log(`${job.uuid} worker started`, job.threadNum); log(`${job.uuid} worker started`, job.threadNum);
try { const object = JSON.parse(job.msg);
const object = JSON.parse(job.msg); let type;
let type; if (object.path) {
if (object.path) { type = object.type;
type = object.type; if (!object.type) {
if (!object.type) { type = await magick.getType(object.path);
type = await magick.getType(object.path);
}
if (!type) {
throw new TypeError("Unknown image type");
}
object.type = type.split("/")[1];
if (object.type !== "gif" && object.onlyGIF) throw new TypeError(`Expected a GIF, got ${object.type}`);
object.delay = object.delay ? object.delay : 0;
} }
if (!type) {
if (object.type === "gif" && !object.delay) { throw new TypeError("Unknown image 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];
} }
object.type = type.split("/")[1];
if (object.type !== "gif" && object.onlyGIF) throw new TypeError(`Expected a GIF, got ${object.type}`);
object.delay = object.delay ? object.delay : 0;
}
const data = await magick.run(object, true); 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];
}
log(`${job.uuid} is done`, job.threadNum); const data = await magick.run(object, true);
const server = net.createServer(function(tcpSocket) {
tcpSocket.write(Buffer.concat([Buffer.from(type ? type : "image/png"), Buffer.from("\n"), data]), (err) => { log(`${job.uuid} is done`, job.threadNum);
if (err) console.error(err); const server = net.createServer(function(tcpSocket) {
tcpSocket.end(() => { tcpSocket.write(Buffer.concat([Buffer.from(type ? type : "image/png"), Buffer.from("\n"), data]), (err) => {
process.exit(); if (err) console.error(err);
}); tcpSocket.end(() => {
process.exit();
}); });
}); });
server.listen(job.port, job.addr); });
// handle address in use errors server.listen(job.port, job.addr);
server.on("error", (e) => { // handle address in use errors
if (e.code === "EADDRINUSE") { server.on("error", (e) => {
console.log("Address in use, retrying..."); if (e.code === "EADDRINUSE") {
setTimeout(() => { console.log("Address in use, retrying...");
server.close(); setTimeout(() => {
server.listen(job.port, job.addr); server.close();
}, 500); server.listen(job.port, job.addr);
} }, 500);
}); }
socket.send(Buffer.concat([Buffer.from([0x1]), Buffer.from(job.uuid), Buffer.from(job.port.toString())]), job.port, job.addr); });
parentPort.postMessage(job.uuid); //Inform main thread about this worker freeing up socket.send(Buffer.concat([Buffer.from([0x1]), Buffer.from(job.uuid), Buffer.from(job.port.toString())]), job.port, job.addr);
} catch (e) { parentPort.postMessage(job.uuid); //Inform main thread about this worker freeing up
socket.send(Buffer.concat([Buffer.from([0x2]), Buffer.from(job.uuid), Buffer.from(e.toString())]), job.port, job.address);
parentPort.postMessage(job.uuid);
}
}); });
} }