151 lines
4.0 KiB
TypeScript
151 lines
4.0 KiB
TypeScript
"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";
|
|
const { MoleculerError } = require("moleculer").Errors;
|
|
const SocketIOService = require("moleculer-io");
|
|
const { v4: uuidv4 } = require("uuid");
|
|
|
|
export default class SocketService extends Service {
|
|
// @ts-ignore
|
|
public constructor(
|
|
public broker: ServiceBroker,
|
|
schema: ServiceSchema<{}> = { name: "socket" }
|
|
) {
|
|
super(broker);
|
|
this.parseServiceSchema({
|
|
name: "socket",
|
|
mixins: [SocketIOService],
|
|
settings: {
|
|
port: process.env.PORT || 3001,
|
|
io: {
|
|
namespaces: {
|
|
"/": {
|
|
events: {
|
|
call: {
|
|
whitelist: ["socket.*"],
|
|
},
|
|
},
|
|
},
|
|
},
|
|
options: {
|
|
adapter: createAdapter(pubClient, subClient),
|
|
},
|
|
},
|
|
},
|
|
hooks: {},
|
|
actions: {
|
|
resumeSession: async (ctx: Context<{ sessionId: string }>) => {
|
|
const { sessionId } = ctx.params;
|
|
console.log("Attempting to resume session...");
|
|
try {
|
|
const sessionRecord = await Session.find({
|
|
sessionId,
|
|
});
|
|
// 1. Check for sessionId's existence, and a match
|
|
if (
|
|
sessionRecord.length !== 0 &&
|
|
sessionRecord[0].sessionId === sessionId
|
|
) {
|
|
// 2. Find if the queue has active, paused or waiting jobs
|
|
const jobs: JobType = await this.broker.call(
|
|
"jobqueue.getJobCountsByType",
|
|
{}
|
|
);
|
|
const { active, paused, waiting } = jobs;
|
|
|
|
if (active > 0 || paused > 0 || waiting > 0) {
|
|
// 3. Get job counts
|
|
const completedJobCount = await pubClient.get(
|
|
"completedJobCount"
|
|
);
|
|
const failedJobCount = await pubClient.get(
|
|
"failedJobCount"
|
|
);
|
|
|
|
// 4. Send the counts to the active socket.io session
|
|
await this.broker.call("socket.broadcast", {
|
|
namespace: "/",
|
|
event: "RESTORE_JOB_COUNTS_AFTER_SESSION_RESTORATION",
|
|
args: [
|
|
{
|
|
completedJobCount,
|
|
failedJobCount,
|
|
queueStatus: "running",
|
|
},
|
|
],
|
|
});
|
|
}
|
|
}
|
|
} catch (err) {
|
|
throw new MoleculerError(
|
|
err,
|
|
500,
|
|
"SESSION_ID_NOT_FOUND",
|
|
{
|
|
data: sessionId,
|
|
}
|
|
);
|
|
}
|
|
},
|
|
|
|
setQueueStatus: async (
|
|
ctx: Context<{
|
|
queueAction: string;
|
|
queueStatus: string;
|
|
}>
|
|
) => {
|
|
const { queueAction } = ctx.params;
|
|
await this.broker.call(
|
|
"jobqueue.toggle",
|
|
{ action: queueAction },
|
|
{}
|
|
);
|
|
},
|
|
importSingleIssue: async (ctx: Context<{}>) => {
|
|
console.info("AirDC++ finished a download -> ");
|
|
console.log(ctx.params);
|
|
// await this.broker.call(
|
|
// "library.importDownloadedComic",
|
|
// { bundle: data },
|
|
// {}
|
|
// );
|
|
},
|
|
},
|
|
methods: {},
|
|
async started() {
|
|
this.io.on("connection", async (socket) => {
|
|
console.log(
|
|
`socket.io server connected to client with session ID: ${socket.id}`
|
|
);
|
|
console.log("Looking up sessionId in Mongo...");
|
|
const sessionIdExists = await Session.find({
|
|
sessionId: socket.handshake.query.sessionId,
|
|
});
|
|
// 1. if sessionId isn't found in Mongo, create one and persist it
|
|
if (sessionIdExists.length === 0) {
|
|
console.log(
|
|
`Socket Id ${socket.id} not found in Mongo, creating a new session...`
|
|
);
|
|
const sessionId = uuidv4();
|
|
socket.sessionId = sessionId;
|
|
console.log(`Saving session ${sessionId} to Mongo...`);
|
|
await Session.create({
|
|
sessionId,
|
|
socketId: socket.id,
|
|
});
|
|
socket.emit("sessionInitialized", sessionId);
|
|
}
|
|
// 2. else, retrieve it from Mongo and "resume" the socket.io connection
|
|
else {
|
|
console.log(`Found socketId ${socket.id}, no-op.`);
|
|
}
|
|
});
|
|
},
|
|
});
|
|
}
|
|
}
|