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

🐰 RabbitMQ for import job first draft

Browse Source
This commit is contained in:
Rishi Ghan
2021-09-13 22:06:19 -07:00
parent 4a39912b2f
commit e9fb359ca3
8 changed files with 353 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

197
package-lock.json generated
View File

@@ -15,6 +15,7 @@
"@types/pino": "^6.3.8",
"@types/string-similarity": "^4.0.0",
"7zip-bin": "^5.1.1",
"amqplib": "^0.7.1",
"fs-extra": "^10.0.0",
"imghash": "^0.0.9",
"leven": "^3.1.0",
@@ -1837,6 +1838,62 @@
"url": "https://github.com/sponsors/epoberezkin"
}
},
"node_modules/amqplib": {
"version": "0.7.1",
"resolved": "https://registry.npmjs.org/amqplib/-/amqplib-0.7.1.tgz",
"integrity": "sha512-KePK3tTOLGU4emTo+PwSDMbc123jrxo13FpRpim1LzJoSlQrIBB2/kMeCC40jK/Zb0olHGaABjLqXDsdK46iLA==",
"dependencies": {
"bitsyntax": "~0.1.0",
"bluebird": "^3.7.2",
"buffer-more-ints": "~1.0.0",
"readable-stream": "1.x >=1.1.9",
"safe-buffer": "~5.2.1",
"url-parse": "~1.5.1"
},
"engines": {
"node": ">=0.8 <=15"
}
},
"node_modules/amqplib/node_modules/isarray": {
"version": "0.0.1",
"resolved": "https://registry.npmjs.org/isarray/-/isarray-0.0.1.tgz",
"integrity": "sha1-ihis/Kmo9Bd+Cav8YDiTmwXR7t8="
},
"node_modules/amqplib/node_modules/readable-stream": {
"version": "1.1.14",
"resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-1.1.14.tgz",
"integrity": "sha1-fPTFTvZI44EwhMY23SB54WbAgdk=",
"dependencies": {
"core-util-is": "~1.0.0",
"inherits": "~2.0.1",
"isarray": "0.0.1",
"string_decoder": "~0.10.x"
}
},
"node_modules/amqplib/node_modules/safe-buffer": {
"version": "5.2.1",
"resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.2.1.tgz",
"integrity": "sha512-rp3So07KcdmmKbGvgaNxQSJr7bGVSVk5S9Eq1F+ppbRo70+YeaDxkw5Dd8NPN+GD6bjnYm2VuPuCXmpuYvmCXQ==",
"funding": [
{
"type": "github",
"url": "https://github.com/sponsors/feross"
},
{
"type": "patreon",
"url": "https://www.patreon.com/feross"
},
{
"type": "consulting",
"url": "https://feross.org/support"
}
]
},
"node_modules/amqplib/node_modules/string_decoder": {
"version": "0.10.31",
"resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-0.10.31.tgz",
"integrity": "sha1-YuIDvEF2bGwoyfyEMB2rHFMQ+pQ="
},
"node_modules/ansi-escapes": {
"version": "4.3.2",
"resolved": "https://registry.npmjs.org/ansi-escapes/-/ansi-escapes-4.3.2.tgz",
@@ -2400,6 +2457,32 @@
"tweetnacl": "^0.14.3"
}
},
"node_modules/bitsyntax": {
"version": "0.1.0",
"resolved": "https://registry.npmjs.org/bitsyntax/-/bitsyntax-0.1.0.tgz",
"integrity": "sha512-ikAdCnrloKmFOugAfxWws89/fPc+nw0OOG1IzIE72uSOg/A3cYptKCjSUhDTuj7fhsJtzkzlv7l3b8PzRHLN0Q==",
"dependencies": {
"buffer-more-ints": "~1.0.0",
"debug": "~2.6.9",
"safe-buffer": "~5.1.2"
},
"engines": {
"node": ">=0.8"
}
},
"node_modules/bitsyntax/node_modules/debug": {
"version": "2.6.9",
"resolved": "https://registry.npmjs.org/debug/-/debug-2.6.9.tgz",
"integrity": "sha512-bC7ElrdJaJnPbAP+1EotYvqZsb3ecl5wi6Bfi6BJTUcNowp6cvspg0jXznRTKDjm/E7AdgFBVeAPVMNcKGsHMA==",
"dependencies": {
"ms": "2.0.0"
}
},
"node_modules/bitsyntax/node_modules/ms": {
"version": "2.0.0",
"resolved": "https://registry.npmjs.org/ms/-/ms-2.0.0.tgz",
"integrity": "sha1-VgiurfwAvmwpAd9fmGF4jeDVl8g="
},
"node_modules/bl": {
"version": "4.1.0",
"resolved": "https://registry.npmjs.org/bl/-/bl-4.1.0.tgz",
@@ -2584,6 +2667,11 @@
"integrity": "sha512-E+XQCRwSbaaiChtv6k6Dwgc+bx+Bs6vuKJHHl5kox/BaKbhiXzqQOwK4cO22yElGp2OCmjwVhT3HmxgyPGnJfQ==",
"dev": true
},
"node_modules/buffer-more-ints": {
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/buffer-more-ints/-/buffer-more-ints-1.0.0.tgz",
"integrity": "sha512-EMetuGFz5SLsT0QTnXzINh4Ksr+oo4i+UGTXEshiGCQWnsgSs7ZhJ8fzlwQ+OzEMs0MpDAMr1hxnblp5a4vcHg=="
},
"node_modules/busboy": {
"version": "0.3.1",
"resolved": "https://registry.npmjs.org/busboy/-/busboy-0.3.1.tgz",
@@ -10064,6 +10152,11 @@
"url": "https://github.com/sponsors/ljharb"
}
},
"node_modules/querystringify": {
"version": "2.2.0",
"resolved": "https://registry.npmjs.org/querystringify/-/querystringify-2.2.0.tgz",
"integrity": "sha512-FIqgj2EUvTa7R50u0rGsyTftzjYmv/a3hO345bZNrqabNqjtgiDMgmo4mkUjd+nzU5oF3dClKqFIPUKybUyqoQ=="
},
"node_modules/quick-format-unescaped": {
"version": "4.0.3",
"resolved": "https://registry.npmjs.org/quick-format-unescaped/-/quick-format-unescaped-4.0.3.tgz",
@@ -10370,6 +10463,11 @@
"integrity": "sha512-NKN5kMDylKuldxYLSUfrbo5Tuzh4hd+2E8NPPX02mZtn1VuREQToYe/ZdlJy+J3uCpfaiGF05e7B8W0iXbQHmg==",
"dev": true
},
"node_modules/requires-port": {
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/requires-port/-/requires-port-1.0.0.tgz",
"integrity": "sha1-kl0mAdOaxIXgkc8NpcbmlNw9yv8="
},
"node_modules/resolve": {
"version": "1.20.0",
"resolved": "https://registry.npmjs.org/resolve/-/resolve-1.20.0.tgz",
@@ -12396,6 +12494,15 @@
"deprecated": "Please see https://github.com/lydell/urix#deprecated",
"dev": true
},
"node_modules/url-parse": {
"version": "1.5.3",
"resolved": "https://registry.npmjs.org/url-parse/-/url-parse-1.5.3.tgz",
"integrity": "sha512-IIORyIQD9rvj0A4CLWsHkBBJuNqWpFQe224b6j9t/ABmquIS0qDU2pY6kl6AuOrL5OkCXHMCFNe1jBcuAggjvQ==",
"dependencies": {
"querystringify": "^2.1.1",
"requires-port": "^1.0.0"
}
},
"node_modules/use": {
"version": "3.1.1",
"resolved": "https://registry.npmjs.org/use/-/use-3.1.1.tgz",
@@ -14326,6 +14433,47 @@
"uri-js": "^4.2.2"
}
},
"amqplib": {
"version": "0.7.1",
"resolved": "https://registry.npmjs.org/amqplib/-/amqplib-0.7.1.tgz",
"integrity": "sha512-KePK3tTOLGU4emTo+PwSDMbc123jrxo13FpRpim1LzJoSlQrIBB2/kMeCC40jK/Zb0olHGaABjLqXDsdK46iLA==",
"requires": {
"bitsyntax": "~0.1.0",
"bluebird": "^3.7.2",
"buffer-more-ints": "~1.0.0",
"readable-stream": "1.x >=1.1.9",
"safe-buffer": "~5.2.1",
"url-parse": "~1.5.1"
},
"dependencies": {
"isarray": {
"version": "0.0.1",
"resolved": "https://registry.npmjs.org/isarray/-/isarray-0.0.1.tgz",
"integrity": "sha1-ihis/Kmo9Bd+Cav8YDiTmwXR7t8="
},
"readable-stream": {
"version": "1.1.14",
"resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-1.1.14.tgz",
"integrity": "sha1-fPTFTvZI44EwhMY23SB54WbAgdk=",
"requires": {
"core-util-is": "~1.0.0",
"inherits": "~2.0.1",
"isarray": "0.0.1",
"string_decoder": "~0.10.x"
}
},
"safe-buffer": {
"version": "5.2.1",
"resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.2.1.tgz",
"integrity": "sha512-rp3So07KcdmmKbGvgaNxQSJr7bGVSVk5S9Eq1F+ppbRo70+YeaDxkw5Dd8NPN+GD6bjnYm2VuPuCXmpuYvmCXQ=="
},
"string_decoder": {
"version": "0.10.31",
"resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-0.10.31.tgz",
"integrity": "sha1-YuIDvEF2bGwoyfyEMB2rHFMQ+pQ="
}
}
},
"ansi-escapes": {
"version": "4.3.2",
"resolved": "https://registry.npmjs.org/ansi-escapes/-/ansi-escapes-4.3.2.tgz",
@@ -14761,6 +14909,31 @@
"tweetnacl": "^0.14.3"
}
},
"bitsyntax": {
"version": "0.1.0",
"resolved": "https://registry.npmjs.org/bitsyntax/-/bitsyntax-0.1.0.tgz",
"integrity": "sha512-ikAdCnrloKmFOugAfxWws89/fPc+nw0OOG1IzIE72uSOg/A3cYptKCjSUhDTuj7fhsJtzkzlv7l3b8PzRHLN0Q==",
"requires": {
"buffer-more-ints": "~1.0.0",
"debug": "~2.6.9",
"safe-buffer": "~5.1.2"
},
"dependencies": {
"debug": {
"version": "2.6.9",
"resolved": "https://registry.npmjs.org/debug/-/debug-2.6.9.tgz",
"integrity": "sha512-bC7ElrdJaJnPbAP+1EotYvqZsb3ecl5wi6Bfi6BJTUcNowp6cvspg0jXznRTKDjm/E7AdgFBVeAPVMNcKGsHMA==",
"requires": {
"ms": "2.0.0"
}
},
"ms": {
"version": "2.0.0",
"resolved": "https://registry.npmjs.org/ms/-/ms-2.0.0.tgz",
"integrity": "sha1-VgiurfwAvmwpAd9fmGF4jeDVl8g="
}
}
},
"bl": {
"version": "4.1.0",
"resolved": "https://registry.npmjs.org/bl/-/bl-4.1.0.tgz",
@@ -14910,6 +15083,11 @@
"integrity": "sha512-E+XQCRwSbaaiChtv6k6Dwgc+bx+Bs6vuKJHHl5kox/BaKbhiXzqQOwK4cO22yElGp2OCmjwVhT3HmxgyPGnJfQ==",
"dev": true
},
"buffer-more-ints": {
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/buffer-more-ints/-/buffer-more-ints-1.0.0.tgz",
"integrity": "sha512-EMetuGFz5SLsT0QTnXzINh4Ksr+oo4i+UGTXEshiGCQWnsgSs7ZhJ8fzlwQ+OzEMs0MpDAMr1hxnblp5a4vcHg=="
},
"busboy": {
"version": "0.3.1",
"resolved": "https://registry.npmjs.org/busboy/-/busboy-0.3.1.tgz",
@@ -20627,6 +20805,11 @@
"side-channel": "^1.0.4"
}
},
"querystringify": {
"version": "2.2.0",
"resolved": "https://registry.npmjs.org/querystringify/-/querystringify-2.2.0.tgz",
"integrity": "sha512-FIqgj2EUvTa7R50u0rGsyTftzjYmv/a3hO345bZNrqabNqjtgiDMgmo4mkUjd+nzU5oF3dClKqFIPUKybUyqoQ=="
},
"quick-format-unescaped": {
"version": "4.0.3",
"resolved": "https://registry.npmjs.org/quick-format-unescaped/-/quick-format-unescaped-4.0.3.tgz",
@@ -20862,6 +21045,11 @@
"integrity": "sha512-NKN5kMDylKuldxYLSUfrbo5Tuzh4hd+2E8NPPX02mZtn1VuREQToYe/ZdlJy+J3uCpfaiGF05e7B8W0iXbQHmg==",
"dev": true
},
"requires-port": {
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/requires-port/-/requires-port-1.0.0.tgz",
"integrity": "sha1-kl0mAdOaxIXgkc8NpcbmlNw9yv8="
},
"resolve": {
"version": "1.20.0",
"resolved": "https://registry.npmjs.org/resolve/-/resolve-1.20.0.tgz",
@@ -22492,6 +22680,15 @@
"integrity": "sha1-2pN/emLiH+wf0Y1Js1wpNQZ6bHI=",
"dev": true
},
"url-parse": {
"version": "1.5.3",
"resolved": "https://registry.npmjs.org/url-parse/-/url-parse-1.5.3.tgz",
"integrity": "sha512-IIORyIQD9rvj0A4CLWsHkBBJuNqWpFQe224b6j9t/ABmquIS0qDU2pY6kl6AuOrL5OkCXHMCFNe1jBcuAggjvQ==",
"requires": {
"querystringify": "^2.1.1",
"requires-port": "^1.0.0"
}
},
"use": {
"version": "3.1.1",
"resolved": "https://registry.npmjs.org/use/-/use-3.1.1.tgz",

1
package.json
View File

@@ -41,6 +41,7 @@
"@types/pino": "^6.3.8",
"@types/string-similarity": "^4.0.0",
"7zip-bin": "^5.1.1",
"amqplib": "^0.7.1",
"fs-extra": "^10.0.0",
"imghash": "^0.0.9",
"leven": "^3.1.0",

35
queue/consumer.ts Normal file
View File

@@ -0,0 +1,35 @@
const amqp = require("amqplib/callback_api");
export const connectQueue = (socketConnection) => {
amqp.connect("amqp://localhost", (error0, connection) => {
if (error0) {
throw error0;
}
connection.createChannel((error1, channel) => {
if (error1) {
throw error1;
}
const queue = "comicBookCovers";
channel.assertQueue(queue, {
durable: false,
});
console.log(
" [*] Waiting for comic book cover data in %s. To exit press CTRL+C",
queue
);
channel.consume(
queue,
(data) => {
console.log("data", data);
//Socket Trigger All Clients
socketConnection.emit("coverExtracted", JSON.parse(data.content.toString()));
},
{
noAck: true,
}
);
});
});
};

34
queue/importQueue.ts Normal file
View File

@@ -0,0 +1,34 @@
import { logger } from "../utils/logger.utils";
//RabbitMQ
const amqp = require("amqplib/callback_api");
const rabbitUrl = "amqp://localhost";
export const sendRabbitMQ = (queueName, data) => {
// connect to local rabbitmq instance
amqp.connect(rabbitUrl, (error0, connection) => {
if (error0) {
throw error0;
}
// create channel
connection.createChannel((error1, channel) => {
if (error1) {
throw error1;
}
channel.prefetch(1);
const queue = queueName;
// Checks for “queueName (updateStock)” queue. If it doesn’t exist, then it creates one.
channel.assertQueue(queue, {
durable: false,
});
channel.sendToQueue(queue, Buffer.from(data));
logger.info(`${data} sent`);
});
setTimeout(function () {
connection.close();
//process.exit(0);
}, 500);
});
};
module.exports = sendRabbitMQ;

81
services/api.service.ts
View File

@@ -1,15 +1,7 @@
import { Service, ServiceBroker, Context } from "moleculer";
import ApiGateway from "moleculer-web";
import { extractCoverFromFile } from "../utils/uncompression.utils";
import { map } from "lodash";
import { connectQueue } from "../queue/consumer";
const IO = require("socket.io")();
import Comic from "../models/comic.model";
import {
IExtractComicBookCoverErrorResponse,
IExtractedComicBookCoverFile,
} from "threetwo-ui-typings";
import { logger } from "../utils/logger.utils";
export default class ApiService extends Service {
public constructor(broker: ServiceBroker) {
super(broker);
@@ -91,77 +83,8 @@ export default class ApiService extends Service {
methods: {},
started(): any {
// Create a Socket.IO instance, passing it our server
this.io = IO.listen(this.server);
// Add a connect listener
this.io.on("connection", (client) => {
this.logger.info("Client connected via websocket!");
client.on(
"importComicsToDB",
async ({ action, params, opts }, done) => {
this.logger.info(
"Received request from client! Action:",
action,
", Params:",
params
);
const { extractionOptions, walkedFolders } = params;
map(walkedFolders, async (folder, idx) => {
let comicExists = await Comic.exists({
"rawFileDetails.name": `${folder.name}`,
});
if (!comicExists) {
let comicBookCoverMetadata:
| IExtractedComicBookCoverFile
| IExtractComicBookCoverErrorResponse
| IExtractedComicBookCoverFile[] = await extractCoverFromFile(
extractionOptions,
folder
);
const dbImportResult =
await this.broker.call(
"import.rawImportToDB",
{
importStatus: {
isImported: true,
tagged: false,
matchedResult: {
score: "0",
},
},
rawFileDetails:
comicBookCoverMetadata,
sourcedMetadata: {
comicvine: {},
},
},
{}
);
client.emit("comicBookCoverMetadata", {
comicBookCoverMetadata,
dbImportResult,
});
} else {
logger.info(
`Comic: \"${folder.name}\" already exists in the database`
);
client.emit("comicBookExists", {
name: folder.name,
});
}
});
}
);
client.on("disconnect", () => {
this.logger.info("Client disconnected");
});
});
connectQueue(this.io);
},
});
}

80
services/import.service.ts
View File

@@ -1,5 +1,5 @@
"use strict";
import { isNil } from "lodash";
import { isNil, map } from "lodash";
import {
Context,
Service,
@@ -12,6 +12,13 @@ import Comic from "../models/comic.model";
import { walkFolder } from "../utils/file.utils";
import { convertXMLToJSON } from "../utils/xml.utils";
import https from "https";
import { logger } from "../utils/logger.utils";
const rabbitmq = require("../queue/importQueue");
import {
IExtractComicBookCoverErrorResponse,
IExtractedComicBookCoverFile,
} from "threetwo-ui-typings";
import { extractCoverFromFile } from "../utils/uncompression.utils";
const ObjectId = require("mongoose").Types.ObjectId;
export default class ImportService extends Service {
@@ -57,6 +64,77 @@ export default class ImportService extends Service {
return convertXMLToJSON("lagos");
},
},
importComicsToDb: {
rest: "POST /importComicsToDB",
params: {},
async handler(
ctx: Context<{
extractionOptions: any;
walkedFolders: [
{
name: string;
extension: string;
containedIn: string;
fileSize: number;
isFile: boolean;
isLink: boolean;
}
];
}>
) {
const { extractionOptions, walkedFolders } =
ctx.params;
map(walkedFolders, async (folder, idx) => {
let comicExists = await Comic.exists({
"rawFileDetails.name": `${folder.name}`,
});
if (!comicExists) {
let comicBookCoverMetadata:
| IExtractedComicBookCoverFile
| IExtractComicBookCoverErrorResponse
| IExtractedComicBookCoverFile[] = await extractCoverFromFile(
extractionOptions,
folder
);
// const dbImportResult =
// await this.broker.call(
// "import.rawImportToDB",
// {
// importStatus: {
// isImported: true,
// tagged: false,
// matchedResult: {
// score: "0",
// },
// },
// rawFileDetails:
// comicBookCoverMetadata,
// sourcedMetadata: {
// comicvine: {},
// },
// },
// {}
// );
rabbitmq(
"comicBookCovers",
JSON.stringify({
comicBookCoverMetadata,
})
);
} else {
logger.info(
`Comic: \"${folder.name}\" already exists in the database`
);
rabbitmq("comicBookCovers",
JSON.stringify({
name: folder.name,
})
);
}
});
},
},
rawImportToDB: {
rest: "POST /rawImportToDB",
params: {},

4
utils/file.utils.ts
View File

@@ -1,6 +1,7 @@
const Walk = require("@root/walk");
import path from "path";
import fs from "fs";
import {
IExplodedPathResponse,
IExtractComicBookCoverErrorResponse,
@@ -21,6 +22,7 @@ export const walkFolder = async (folder: string): Promise<IFolderData[]> => {
containedIn: "",
isFile: false,
isLink: true,
fileSize: 0,
};
const walk = Walk.create({ sort: filterOutDotFiles });
@@ -30,9 +32,11 @@ export const walkFolder = async (folder: string): Promise<IFolderData[]> => {
return false;
}
if ([".cbz", ".cbr"].includes(path.extname(dirent.name))) {
console.log(path.resolve(pathname));
walkResult = {
name: path.basename(dirent.name, path.extname(dirent.name)),
extension: path.extname(dirent.name),
fileSize: fs.statSync(path.resolve(pathname)).size,
containedIn: path.dirname(pathname),
isFile: dirent.isFile(),
isLink: dirent.isSymbolicLink(),

2
utils/uncompression.utils.ts
View File

@@ -97,7 +97,7 @@ export const extractCoverFromFile = async (
resolve({
name: walkedFolder.name,
path: renditionPath,
fileSize: stats.size,
fileSize: walkedFolder.fileSize,
extension: path.extname(constructedPaths.inputFilePath),
containedIn: walkedFolder.containedIn,
calibreMetadata: {