Merge branch 'master' into getbundles-fix
This commit is contained in:
GitHub
@@ -14,7 +14,6 @@ import { pubClient } from "../config/redis.config";
|
||||
import path from "path";
|
||||
const { MoleculerError } = require("moleculer").Errors;
|
||||
|
||||
console.log(process.env.REDIS_URI);
|
||||
export default class JobQueueService extends Service {
|
||||
public constructor(public broker: ServiceBroker) {
|
||||
super(broker);
|
||||
@@ -22,9 +21,10 @@ export default class JobQueueService extends Service {
|
||||
name: "jobqueue",
|
||||
hooks: {},
|
||||
mixins: [DbMixin("comics", Comic), BullMqMixin],
|
||||
|
||||
settings: {
|
||||
bullmq: {
|
||||
client: process.env.REDIS_URI,
|
||||
client: pubClient,
|
||||
},
|
||||
},
|
||||
actions: {
|
||||
@@ -57,20 +57,24 @@ export default class JobQueueService extends Service {
|
||||
handler: async (
|
||||
ctx: Context<{ action: string; description: string }>
|
||||
) => {
|
||||
const { action, description } = ctx.params;
|
||||
// Enqueue the job
|
||||
const job = await this.localQueue(
|
||||
ctx,
|
||||
action,
|
||||
ctx.params,
|
||||
{
|
||||
priority: 10,
|
||||
}
|
||||
);
|
||||
console.log(`Job ${job.id} enqueued`);
|
||||
console.log(`${description}`);
|
||||
try {
|
||||
const { action, description } = ctx.params;
|
||||
// Enqueue the job
|
||||
const job = await this.localQueue(
|
||||
ctx,
|
||||
action,
|
||||
{},
|
||||
{
|
||||
priority: 10,
|
||||
}
|
||||
);
|
||||
console.log(`Job ${job.id} enqueued`);
|
||||
console.log(`${description}`);
|
||||
|
||||
return job.id;
|
||||
return job.id;
|
||||
} catch (error) {
|
||||
console.error("Failed to enqueue job:", error);
|
||||
}
|
||||
},
|
||||
},
|
||||
|
||||
|
||||
@@ -59,9 +59,11 @@ import path from "path";
|
||||
import { COMICS_DIRECTORY, USERDATA_DIRECTORY } from "../constants/directories";
|
||||
import AirDCPPSocket from "../shared/airdcpp.socket";
|
||||
|
||||
console.log(`MONGO -> ${process.env.MONGO_URI}`);
|
||||
export default class ImportService extends Service {
|
||||
public constructor(public broker: ServiceBroker) {
|
||||
export default class LibraryService extends Service {
|
||||
public constructor(
|
||||
public broker: ServiceBroker,
|
||||
schema: ServiceSchema<{}> = { name: "library" }
|
||||
) {
|
||||
super(broker);
|
||||
this.parseServiceSchema({
|
||||
name: "library",
|
||||
@@ -164,78 +166,52 @@ export default class ImportService extends Service {
|
||||
},
|
||||
newImport: {
|
||||
rest: "POST /newImport",
|
||||
// params: {},
|
||||
async handler(
|
||||
ctx: Context<{
|
||||
extractionOptions?: any;
|
||||
sessionId: string;
|
||||
}>
|
||||
) {
|
||||
async handler(ctx) {
|
||||
const { sessionId } = ctx.params;
|
||||
try {
|
||||
// Get params to be passed to the import jobs
|
||||
const { sessionId } = ctx.params;
|
||||
// 1. Walk the Source folder
|
||||
klaw(path.resolve(COMICS_DIRECTORY))
|
||||
// 1.1 Filter on .cb* extensions
|
||||
.pipe(
|
||||
through2.obj(function (item, enc, next) {
|
||||
let fileExtension = path.extname(
|
||||
item.path
|
||||
);
|
||||
if (
|
||||
[".cbz", ".cbr", ".cb7"].includes(
|
||||
fileExtension
|
||||
)
|
||||
) {
|
||||
this.push(item);
|
||||
}
|
||||
next();
|
||||
})
|
||||
)
|
||||
// 1.2 Pipe filtered results to the next step
|
||||
// Enqueue the job in the queue
|
||||
.on("data", async (item) => {
|
||||
console.info(
|
||||
"Found a file at path: %s",
|
||||
item.path
|
||||
);
|
||||
let comicExists = await Comic.exists({
|
||||
"rawFileDetails.name": `${path.basename(
|
||||
item.path,
|
||||
path.extname(item.path)
|
||||
)}`,
|
||||
});
|
||||
if (!comicExists) {
|
||||
// 2.1 Reset the job counters in Redis
|
||||
await pubClient.set(
|
||||
"completedJobCount",
|
||||
0
|
||||
);
|
||||
await pubClient.set(
|
||||
"failedJobCount",
|
||||
0
|
||||
);
|
||||
// 2.2 Send the extraction job to the queue
|
||||
this.broker.call("jobqueue.enqueue", {
|
||||
fileObject: {
|
||||
filePath: item.path,
|
||||
fileSize: item.stats.size,
|
||||
},
|
||||
sessionId,
|
||||
importType: "new",
|
||||
action: "enqueue.async",
|
||||
});
|
||||
} else {
|
||||
console.log(
|
||||
"Comic already exists in the library."
|
||||
);
|
||||
}
|
||||
})
|
||||
.on("end", () => {
|
||||
console.log("All files traversed.");
|
||||
// Initialize Redis counters once at the start of the import
|
||||
await pubClient.set("completedJobCount", 0);
|
||||
await pubClient.set("failedJobCount", 0);
|
||||
|
||||
// Convert klaw to use a promise-based approach for better flow control
|
||||
const files = await this.getComicFiles(
|
||||
COMICS_DIRECTORY
|
||||
);
|
||||
for (const file of files) {
|
||||
console.info(
|
||||
"Found a file at path:",
|
||||
file.path
|
||||
);
|
||||
const comicExists = await Comic.exists({
|
||||
"rawFileDetails.name": path.basename(
|
||||
file.path,
|
||||
path.extname(file.path)
|
||||
),
|
||||
});
|
||||
|
||||
if (!comicExists) {
|
||||
// Send the extraction job to the queue
|
||||
await this.broker.call("jobqueue.enqueue", {
|
||||
fileObject: {
|
||||
filePath: file.path,
|
||||
fileSize: file.stats.size,
|
||||
},
|
||||
sessionId,
|
||||
importType: "new",
|
||||
action: "enqueue.async",
|
||||
});
|
||||
} else {
|
||||
console.log(
|
||||
"Comic already exists in the library."
|
||||
);
|
||||
}
|
||||
}
|
||||
console.log("All files traversed.");
|
||||
} catch (error) {
|
||||
console.log(error);
|
||||
console.error(
|
||||
"Error during newImport processing:",
|
||||
error
|
||||
);
|
||||
}
|
||||
},
|
||||
},
|
||||
@@ -864,7 +840,35 @@ export default class ImportService extends Service {
|
||||
},
|
||||
},
|
||||
},
|
||||
methods: {},
|
||||
methods: {
|
||||
// Method to walk the directory and filter comic files
|
||||
getComicFiles: (directory) => {
|
||||
return new Promise((resolve, reject) => {
|
||||
const files = [];
|
||||
klaw(directory)
|
||||
.pipe(
|
||||
through2.obj(function (item, enc, next) {
|
||||
const fileExtension = path.extname(
|
||||
item.path
|
||||
);
|
||||
if (
|
||||
[".cbz", ".cbr", ".cb7"].includes(
|
||||
fileExtension
|
||||
)
|
||||
) {
|
||||
this.push(item);
|
||||
}
|
||||
next();
|
||||
})
|
||||
)
|
||||
.on("data", (item) => {
|
||||
files.push(item);
|
||||
})
|
||||
.on("end", () => resolve(files))
|
||||
.on("error", (err) => reject(err));
|
||||
});
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
"use strict";
|
||||
import { Service, ServiceBroker, ServiceSchema, Context } from "moleculer";
|
||||
import { JobType } from "moleculer-bullmq";
|
||||
import { createClient } from "redis";
|
||||
import { createAdapter } from "@socket.io/redis-adapter";
|
||||
import Session from "../models/session.model";
|
||||
import { pubClient, subClient } from "../config/redis.config";
|
||||
@@ -274,6 +273,22 @@ export default class SocketService extends Service {
|
||||
},
|
||||
},
|
||||
async started() {
|
||||
this.logger.info("Starting Socket Service...");
|
||||
this.logger.debug("pubClient:", pubClient);
|
||||
this.logger.debug("subClient:", subClient);
|
||||
if (!pubClient || !subClient) {
|
||||
this.logger.error("Redis clients are not initialized!");
|
||||
throw new Error("Redis clients are not initialized!");
|
||||
}
|
||||
|
||||
// Additional checks or logic if necessary
|
||||
if (pubClient.status !== "ready") {
|
||||
await pubClient.connect();
|
||||
}
|
||||
|
||||
if (subClient.status !== "ready") {
|
||||
await subClient.connect();
|
||||
}
|
||||
this.io.on("connection", async (socket) => {
|
||||
console.log(
|
||||
`socket.io server connected to client with session ID: ${socket.id}`
|
||||
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
import { DbMixin } from "../mixins/db.mixin";
|
||||
import Comic from "../models/comic.model";
|
||||
import BullMqMixin from "moleculer-bullmq";
|
||||
import { pubClient } from "../config/redis.config";
|
||||
const { MoleculerError } = require("moleculer").Errors;
|
||||
|
||||
export default class ImageTransformation extends Service {
|
||||
@@ -23,7 +24,7 @@ export default class ImageTransformation extends Service {
|
||||
mixins: [DbMixin("comics", Comic), BullMqMixin],
|
||||
settings: {
|
||||
bullmq: {
|
||||
client: process.env.REDIS_URI,
|
||||
client: pubClient,
|
||||
},
|
||||
},
|
||||
hooks: {},
|
||||
|
||||
Reference in New Issue
Block a user