use redis service
This commit is contained in:
1 parent
0687f737a7
commit
0694bd06d1
3 files changed
+24
-22
No files matched your search
+17
-12
@@ -81,7 +81,9 @@ import ServiceRegistry from "./service/serviceRegistry.js";
|
||||
|
||||
import MongoDB from "./db/mongo/MongoDB.js";
|
||||
|
||||
// Redis Service and dependencies
|
||||
import IORedis from "ioredis";
|
||||
import RedisService from "./service/redisService.js";
|
||||
|
||||
import TranslationService from "./service/translationService.js";
|
||||
import languageMiddleware from "./middleware/languageMiddleware.js";
|
||||
@@ -113,21 +115,16 @@ const shutdown = async () => {
|
||||
method: "shutdown",
|
||||
});
|
||||
// flush Redis
|
||||
const settings =
|
||||
ServiceRegistry.get(SettingsService.SERVICE_NAME).getSettings() || {};
|
||||
const redisService = ServiceRegistry.get(RedisService.SERVICE_NAME);
|
||||
await redisService.flushall();
|
||||
|
||||
const { redisUrl } = settings;
|
||||
const redis = new IORedis(redisUrl, { maxRetriesPerRequest: null });
|
||||
|
||||
logger.info({ message: "Flushing Redis" });
|
||||
await redis.flushall();
|
||||
logger.info({ message: "Redis flushed" });
|
||||
process.exit(1);
|
||||
}, SHUTDOWN_TIMEOUT);
|
||||
try {
|
||||
server.close();
|
||||
await ServiceRegistry.get(JobQueue.SERVICE_NAME).obliterate();
|
||||
await ServiceRegistry.get(MongoDB.SERVICE_NAME).disconnect();
|
||||
await ServiceRegistry.get(RedisService.SERVICE_NAME).flushall();
|
||||
logger.info({ message: "Graceful shutdown complete" });
|
||||
process.exit(0);
|
||||
} catch (error) {
|
||||
@@ -185,7 +182,13 @@ const startApp = async () => {
|
||||
stringService
|
||||
);
|
||||
|
||||
const jobQueue = new JobQueue(
|
||||
const redisService = await RedisService.createInstance({
|
||||
logger,
|
||||
IORedis,
|
||||
SettingsService: settingsService,
|
||||
});
|
||||
|
||||
const jobQueue = new JobQueue({
|
||||
db,
|
||||
statusService,
|
||||
networkService,
|
||||
@@ -194,8 +197,9 @@ const startApp = async () => {
|
||||
stringService,
|
||||
logger,
|
||||
Queue,
|
||||
Worker
|
||||
);
|
||||
Worker,
|
||||
redisService,
|
||||
});
|
||||
|
||||
// Register services
|
||||
ServiceRegistry.register(JobQueue.SERVICE_NAME, jobQueue);
|
||||
@@ -207,6 +211,7 @@ const startApp = async () => {
|
||||
ServiceRegistry.register(StatusService.SERVICE_NAME, statusService);
|
||||
ServiceRegistry.register(NotificationService.SERVICE_NAME, notificationService);
|
||||
ServiceRegistry.register(TranslationService.SERVICE_NAME, translationService);
|
||||
ServiceRegistry.register(RedisService.SERVICE_NAME, redisService);
|
||||
|
||||
await translationService.initialize();
|
||||
|
||||
@@ -343,7 +348,7 @@ const startApp = async () => {
|
||||
app.use("/api/v1/queue", verifyJWT, queueRoutes.getRouter());
|
||||
app.use("/api/v1/distributed-uptime", distributedUptimeRoutes.getRouter());
|
||||
app.use("/api/v1/status-page", statusPageRoutes.getRouter());
|
||||
app.use("/api/v1/notifications", verifyJWT, notificationRoutes.getRouter());
|
||||
app.use("/api/v1/notifications", verifyJWT, notificationRoutes.getRouter());
|
||||
app.use("/api/v1/diagnostic", verifyJWT, diagnosticRoutes.getRouter());
|
||||
app.use("/api/v1/health", (req, res) => {
|
||||
res.json({
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
import IORedis from "ioredis";
|
||||
|
||||
const QUEUE_NAMES = ["uptime", "pagespeed", "hardware", "distributed"];
|
||||
const SERVICE_NAME = "JobQueue";
|
||||
const JOBS_PER_WORKER = 5;
|
||||
@@ -18,7 +16,7 @@ const getSchedulerId = (monitor) => `scheduler:${monitor.type}:${monitor._id}`;
|
||||
class NewJobQueue {
|
||||
static SERVICE_NAME = SERVICE_NAME;
|
||||
|
||||
constructor(
|
||||
constructor({
|
||||
db,
|
||||
statusService,
|
||||
networkService,
|
||||
@@ -27,16 +25,14 @@ class NewJobQueue {
|
||||
stringService,
|
||||
logger,
|
||||
Queue,
|
||||
Worker
|
||||
) {
|
||||
const settings = settingsService.getSettings() || {};
|
||||
const { redisUrl } = settings;
|
||||
const connection = new IORedis(redisUrl, { maxRetriesPerRequest: null });
|
||||
Worker,
|
||||
redisService,
|
||||
}) {
|
||||
this.connection = redisService.getConnection();
|
||||
this.queues = {};
|
||||
this.workers = {};
|
||||
this.lastJobProcessedTime = {};
|
||||
|
||||
this.connection = connection;
|
||||
this.db = db;
|
||||
this.networkService = networkService;
|
||||
this.statusService = statusService;
|
||||
@@ -47,7 +43,7 @@ class NewJobQueue {
|
||||
this.stringService = stringService;
|
||||
|
||||
QUEUE_NAMES.forEach((name) => {
|
||||
const q = new Queue(name, { connection });
|
||||
const q = new Queue(name, { connection: this.connection });
|
||||
this.lastJobProcessedTime[q.name] = Date.now();
|
||||
q.on("error", (error) => {
|
||||
this.logger.error({
|
||||
|
||||
@@ -31,6 +31,7 @@ class RedisService {
|
||||
return Math.min(times * 100, 2000);
|
||||
},
|
||||
});
|
||||
|
||||
await new Promise((resolve, reject) => {
|
||||
let errorOccurred = false;
|
||||
|
||||
|
||||
Reference in new issue
Block a user