feat: add workerExit event & add support to communication between 2 workers and not all

Author: henriquemdimer

src/core/cluster/ClusterManager.ts

@@ -3,7 +3,6 @@ import { ClientOptions } from '../Client';
 import Cluster, { Worker } from 'cluster';
 import { IPCMessage, IResult } from '../classes/IPCMessage';
 import { ClusterClient } from './ClusterClient';
-import { Observable } from '../classes/Observable';
 import { ClusterEvents } from '../events/ClusterEvents';
 
 export interface ClusterManagerOptions extends ClientOptions {
@@ -49,7 +48,12 @@ export class ClusterManager {
         this.workers.push(worker);
 
         worker.send({ index: i, token: this.#token, shards: chunk, total: this.options.sharding?.totalShards || 2, event: 'master-initial' });
-
+        
+        worker.on('exit', () => {
+          const index = this.workers.findIndex((w) => w.process.pid == worker.process.pid);
+          this.workers.splice(index, 1);
+          this.events.workerExit.notify(worker);
+        });
         worker.on('message', (message) => {
           const msg = new IPCMessage(message);
           if (!msg) return;
@@ -83,6 +87,25 @@ export class ClusterManager {
 
             break;
           }
+
+          case 'data-request': {
+            const target = this.workers.find((w) => w.process.pid == msg.to);
+            if(target) {
+              target.send(new IPCMessage({ from: msg.from, to: target.process.pid, event: msg.event, cid: msg.cid, data: msg.data }));
+              let result: IResult;
+
+              const callback = (message: any) => {
+                const response = new IPCMessage(message);
+                if(!response || response.cid != msg.cid || !response.result) return;
+
+                result = response.result;
+                target.removeListener('message', callback);
+                sender.send(new IPCMessage({ from: msg.to || '', to: msg.from, event: 'data-response', cid: msg.cid, data: result }));
+              };
+
+              target.on('message', callback);
+            }
+          }
           }
         });
       }

src/core/cluster/IPC.ts

@@ -23,7 +23,7 @@ export class IPC extends EventEmitter {
       case 'data-request': {
         if (!msg.data) return;
 
-        const context = { client: this.#client };
+        const context = { client: this.#client, process };
         VM.createContext(context);
 
         const result = VM.runInContext(msg.data, context);
@@ -58,4 +58,8 @@ export class IPC extends EventEmitter {
       process.on('message', callback);
     });
   }
+
+  public generateCID() {
+    return IPCMessage.generateCID();
+  }
 }

src/core/events/ClusterEvents.ts

@@ -1,6 +1,8 @@
+import { Worker } from 'cluster';
 import { Observable } from '../classes/Observable';
 import { ClusterClient } from '../cluster/ClusterClient';
 
 export class ClusterEvents {
   public workerSpawned = new Observable<ClusterClient>();
+  public workerExit = new Observable<Worker>();
 }

src/test/cluster.ts

@@ -17,7 +17,8 @@ import { Intents } from '../';
     intents: [Intents.ALL]
   });
 
-  (await manager.init()).events.workerSpawned.subscribe((client: ClusterClient) => {
+  await manager.init();
+  manager.events.workerSpawned.subscribe((client: ClusterClient) => {
     client.events.ready.subscribe(() => {
       console.log(client.user?.username, 'is ready', 'PID =', process.pid);
     });
@@ -30,4 +31,8 @@ import { Intents } from '../';
 
     client.init();
   });
+
+  manager.events.workerExit.subscribe((worker) => {
+    console.log('Worker died:', worker.process.pid);
+  });
 })();