Workers defined for Redis Subs

kafka-clickhouse/package.json

@@ -24,7 +24,7 @@
   },
   "scripts": {
     "dev": "nodemon --env-file=.env build/src/main.js",
-    "kafka:worker": "nodemon --env-file=.env build/src/services/redis/subscriber.js",
+    "kafka:worker": "nodemon --env-file=.env build/src/services/redis/worker.js",
     "start": "node build/src/main.js",
     "clean": "rimraf coverage build tmp",
     "prebuild": "npm run lint",

kafka-clickhouse/src/main.ts

@@ -2,6 +2,7 @@ import express, { Application } from 'express';
 
 import { secretKeyValidator } from './middleware.js';
 import router from './router.js';
+import log from './services/redis/worker.js'
 
 const app: Application = express();
 
@@ -15,8 +16,12 @@ app.get('/', async (_req, res) => {
   });
 });
 
+
+
 app.use('/api', router);
 
+console.log(log());
+
 const PORT = process.env.PORT;
 app.listen(PORT, () => {
   console.log(`Server is listening on port ${PORT}`);

kafka-clickhouse/src/services/redis/connection.ts

@@ -1,6 +1,7 @@
 import { Redis } from 'ioredis';
 import {UserDeviceInfo} from './types.js'
 
+
 // Create a Redis client instance
 const redis = new Redis(process.env.REDIS_URI);
 

kafka-clickhouse/src/services/redis/subscriber.ts

@@ -1,3 +1,5 @@
+// worker.js
+import { parentPort } from 'worker_threads';
 import RedisQueue from './connection.js';
 import KafkaProducer from '../kafka/producer.js';
 
@@ -33,3 +35,11 @@ console.log('Queue processor started. Waiting for messages...');
 processQueue().catch((error) => {
   console.error('Error starting queue processor:', error);
 });
+
+if (parentPort) {
+  parentPort.on('message', (message) => {
+    if (message === 'stop') {
+      running = false;
+    }
+  });
+}

kafka-clickhouse/src/services/redis/worker.ts

@@ -0,0 +1,50 @@
+import { Worker } from 'worker_threads';
+
+const numWorkers = 2;
+const workers = [];
+
+import path from 'path';
+
+const __dirname = path.resolve();
+
+function getDynamicPath(): string {
+    const loc = path.resolve(__dirname, 'build/src/services/redis/subscriber.js');
+    return loc;
+  }
+
+for (let i = 0; i < numWorkers; i++) {
+  const worker = new Worker(getDynamicPath());
+  workers.push(worker);
+
+  worker.on('exit', (code) => {
+    if (code !== 0) {
+      console.error(`Worker stopped with exit code ${code}`);
+    }
+  });
+
+  worker.on('error', (error) => {
+    console.error('Worker error:', error);
+  });
+}
+
+process.on('SIGINT', () => {
+  console.log('Received SIGINT. Shutting down...');
+  stopWorkers();
+});
+
+process.on('SIGTERM', () => {
+  console.log('Received SIGTERM. Shutting down...');
+  stopWorkers();
+});
+
+function stopWorkers(): void {
+  for (const worker of workers) {
+    worker.postMessage('stop');
+  }
+}
+
+function log(): string {
+  return 'Worker threads started';
+}
+
+export default log;