frishi/threetwo-core-service
1
0
Fork 0
Code Issues 2 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

🔌 Added a socket action emit for uncompression/resize

Browse Source
This commit is contained in:
Rishi Ghan
2022-12-09 12:17:22 -08:00
parent 9a41f34099
commit fe4ba87256
1 changed files with 23 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
services/importqueue.service.ts
View File

@@ -38,7 +38,10 @@ import { Context, Service, ServiceBroker, ServiceSchema } from "moleculer";
import BullMQMixin, { SandboxedJob } from "moleculer-bull"; import BullMQMixin, { SandboxedJob } from "moleculer-bull";
import { DbMixin } from "../mixins/db.mixin"; import { DbMixin } from "../mixins/db.mixin";
import Comic from "../models/comic.model"; import Comic from "../models/comic.model";
import { extractFromArchive, uncompressEntireArchive } from "../utils/uncompression.utils"; import {
extractFromArchive,
uncompressEntireArchive,
} from "../utils/uncompression.utils";
const REDIS_URI = process.env.REDIS_URI || `redis://localhost:6379`; const REDIS_URI = process.env.REDIS_URI || `redis://localhost:6379`;
const EventEmitter = require("events"); const EventEmitter = require("events");
@@ -145,8 +148,11 @@ export default class QueueService extends Service {
"process.uncompressAndResize": { "process.uncompressAndResize": {
concurrency: 2, concurrency: 2,
async process(job: SandboxedJob) { async process(job: SandboxedJob) {
console.log(``); console.log(`Initiating uncompression job...`);
return await uncompressEntireArchive(job.data.filePath, job.data.options); return await uncompressEntireArchive(
job.data.filePath,
job.data.options
);
}, },
}, },
}, },
@@ -220,15 +226,15 @@ export default class QueueService extends Service {
event: "action", event: "action",
args: [{ type: "LS_COVER_EXTRACTED", result: res }], //optional args: [{ type: "LS_COVER_EXTRACTED", result: res }], //optional
}); });
console.info(`Import Job with the id '${job.id}' completed.`); console.info(
`Import Job with the id '${job.id}' completed.`
);
} }
); );
await this.getQueue("process.import").on( await this.getQueue("process.import").on(
"stalled", "stalled",
async (job) => { async (job) => {
console.warn( console.warn(`Import job '${job.id} stalled!`);
`Import job '${job.id} stalled!`
);
console.log(`${JSON.stringify(job, null, 2)}`); console.log(`${JSON.stringify(job, null, 2)}`);
console.log(`is stalled.`); console.log(`is stalled.`);
} }
@@ -240,9 +246,17 @@ export default class QueueService extends Service {
await this.broker.call("socket.broadcast", { await this.broker.call("socket.broadcast", {
namespace: "/", namespace: "/",
event: "action", event: "action",
args: [{ type: "COMICBOOK_EXTRACTION_SUCCESS", result: res }] args: [
{
type: "COMICBOOK_EXTRACTION_SUCCESS",
result: {
files: res,
purpose: job.data.options.purpose,
},
},
],
}); });
console.info(`Uncompression Job ${job.id} completed.`) console.info(`Uncompression Job ${job.id} completed.`);
} }
); );
}, },