feat(cluster): add peer syncing

When adding a new node all peers will be broadcasted now, which should
lead to remote nodes picking it up quickly as well.

src/cluster/cluster.service.ts

@@ -2,7 +2,8 @@ import {
   Injectable,
   Logger,
   OnApplicationBootstrap,
-  OnApplicationShutdown
+  OnApplicationShutdown,
+  OnModuleInit
 } from '@nestjs/common';
 import Democracy, { Node } from 'democracy';
 import { Advertisement, Browser, Service } from 'mdns';
@@ -21,7 +22,7 @@ try {
 
 @Injectable()
 export class ClusterService extends Democracy
-  implements OnApplicationBootstrap, OnApplicationShutdown {
+  implements OnModuleInit, OnApplicationBootstrap, OnApplicationShutdown {
   private readonly configService: ConfigService;
   private readonly config: ClusterConfig;
   private readonly logger: Logger;
@@ -54,6 +55,13 @@ export class ClusterService extends Democracy
     this.logger = new Logger(ClusterService.name);
   }
 
+  onModuleInit(): void {
+    this.on('added', this.handleNodeAdded);
+    this.on('removed', this.handleNodeRemoved);
+    this.on('elected', this.handleNodeElected);
+    this.on('peers', this.addMissingPeers);
+  }
+
   onApplicationBootstrap(): void {
     if (this.config.autoDiscovery) {
       if (mdns !== undefined) {
@@ -65,9 +73,7 @@ export class ClusterService extends Democracy
       }
     }
 
-    this.on('added', this.handleNodeAdded);
-    this.on('removed', this.handleNodeRemoved);
-    this.on('elected', this.handleNodeElected);
+    this.broadcastPeers();
   }
 
   onApplicationShutdown(): void {
@@ -86,6 +92,22 @@ export class ClusterService extends Democracy
     return !this.config.quorum || activeNodes.length >= this.config.quorum;
   }
 
+  broadcastPeers(): void {
+    this.send('peers', this.options.peers);
+  }
+
+  protected addMissingPeers(peers: string[][]): void {
+    peers.forEach(peer => {
+      const index = this.options.peers.findIndex(
+        p => p[0] === peer[0] && p[1] === peer[1]
+      );
+
+      if (index < 0) {
+        this.options.peers.push(peer);
+      }
+    });
+  }
+
   protected startBonjourDiscovery(): void {
     this.advertisement = mdns.createAdvertisement(
       mdns.udp('room-assistant'),
@@ -128,11 +150,12 @@ export class ClusterService extends Democracy
     }
 
     this.options.peers.push([service.addresses[0], service.port.toString()]);
-    this.hello();
+    this.send('hello');
   }
 
   private handleNodeAdded(node: Node): void {
     this.logger.log(`Added ${node.source} to the cluster with id ${node.id}`);
+    this.broadcastPeers();
   }
 
   private handleNodeRemoved(node: Node): void {

typings/democracy/index.d.ts

@@ -1,6 +1,11 @@
 declare module 'democracy' {
+  import {Socket} from 'dgram';
   export default class Democracy {
     protected options: InternalOptions;
+    protected socket: Socket;
+    protected _id: string;
+    protected _weight: number;
+    protected _state: 'leader' | 'citizen' | 'removed';
 
     constructor(opts?: Options);
     protected addNodeToList(node: Node): void;
@@ -9,7 +14,7 @@ declare module 'democracy' {
     leader(): Node;
     resign(): this;
     isLeader(): boolean;
-    send(customEvent: string, extraData: any, id?: string): this;
+    send(customEvent: string, extraData?: any, id?: string): this;
     subscribe(channel: string): this;
     publish(channel: string, msg: any): this;
     on(event: string, func: (data: any) => void);