Categorize nodes by chains

packages/backend/src/Aggregator.ts

@@ -0,0 +1,84 @@
+import Chain from './Chain';
+import Node from './Node';
+import Feed from './Feed';
+import FeedSet from './FeedSet';
+import { Types, FeedMessage } from '@dotstats/common';
+
+export default class Aggregator {
+    private readonly chains = new Map<Types.ChainLabel, Chain>();
+    private readonly feeds = new FeedSet();
+
+    constructor() {
+        setInterval(() => this.timeoutCheck(), 10000);
+    }
+
+    public addNode(node: Node) {
+        let chain = this.getChain(node.chain);
+
+        chain.addNode(node);
+    }
+
+    public addFeed(feed: Feed) {
+        this.feeds.add(feed);
+
+        for (const chain of this.chains.values()) {
+            feed.sendMessage(Feed.addedChain(chain.label));
+        }
+
+        feed.events.on('subscribe', (label: Types.ChainLabel) => {
+            const chain = this.chains.get(label);
+
+            if (chain) {
+                chain.addFeed(feed);
+                feed.sendMessage(Feed.subscribedTo(label));
+            }
+        })
+
+        feed.events.on('unsubscribe', (label: Types.ChainLabel) => {
+            const chain = this.chains.get(label);
+
+            if (chain) {
+                chain.removeFeed(feed);
+                feed.sendMessage(Feed.unsubscribedFrom(label));
+            }
+        });
+    }
+
+    private getChain(label: Types.ChainLabel): Chain {
+        const chain = this.chains.get(label);
+
+        if (chain) {
+            return chain;
+        } else {
+            const chain = new Chain(label);
+
+            chain.events.on('disconnect', (count: number) => {
+                if (count !== 0) {
+                    return;
+                }
+
+                chain.events.removeAllListeners();
+
+                this.chains.delete(chain.label);
+
+                console.log(`Chain: ${label} lost all nodes`);
+                this.feeds.broadcast(Feed.removedChain(label));
+            });
+
+            this.chains.set(label, chain);
+
+            console.log(`New chain: ${label}`);
+            this.feeds.broadcast(Feed.addedChain(label));
+
+            return chain;
+        }
+    }
+
+    private timeoutCheck() {
+        const empty: Types.ChainLabel[] = [];
+
+        for (const chain of this.chains.values()) {
+            chain.timeoutCheck();
+        }
+    }
+}

packages/backend/src/Chain.ts

@@ -0,0 +1,90 @@
+import * as EventEmitter from 'events';
+import Node from './Node';
+import Feed from './Feed';
+import FeedSet from './FeedSet';
+import { timestamp, Types, FeedMessage } from '@dotstats/common';
+
+export default class Chain {
+    private nodes = new Set<Node>();
+    private feeds = new FeedSet();
+
+    public readonly events = new EventEmitter();
+    public readonly label: Types.ChainLabel;
+
+    public height = 0 as Types.BlockNumber;
+    public blockTimestamp = 0 as Types.Timestamp;
+
+    constructor(label: Types.ChainLabel) {
+        this.label = label;
+    }
+
+    public get nodeCount(): number {
+        return this.nodes.size;
+    }
+
+    public addNode(node: Node) {
+        console.log(`[${this.label}] new node: ${node.name}`);
+
+        this.nodes.add(node);
+        this.feeds.broadcast(Feed.addedNode(node));
+
+        node.events.once('disconnect', () => {
+            node.events.removeAllListeners();
+
+            this.nodes.delete(node);
+            this.feeds.broadcast(Feed.removedNode(node));
+
+            this.events.emit('disconnect', this.nodeCount);
+        });
+
+        node.events.on('block', () => this.updateBlock(node));
+        node.events.on('stats', () => this.feeds.broadcast(Feed.stats(node)));
+    }
+
+    public addFeed(feed: Feed) {
+        this.feeds.add(feed);
+
+        // TODO: this is a bit unclean, find a better way
+        feed.chain = this.label;
+
+        feed.sendMessage(Feed.timeSync());
+        feed.sendMessage(Feed.bestBlock(this.height, this.blockTimestamp));
+
+        for (const node of this.nodes.values()) {
+            feed.sendMessage(Feed.addedNode(node));
+        }
+    }
+
+    public removeFeed(feed: Feed) {
+        this.feeds.remove(feed);
+    }
+
+    public nodeList(): IterableIterator<Node> {
+        return this.nodes.values();
+    }
+
+    public timeoutCheck() {
+        const now = timestamp();
+
+        for (const node of this.nodes.values()) {
+            node.timeoutCheck(now);
+        }
+
+        this.feeds.broadcast(Feed.timeSync());
+    }
+
+    private updateBlock(node: Node) {
+        if (node.height > this.height) {
+            this.height = node.height;
+            this.blockTimestamp = node.blockTimestamp;
+
+            this.feeds.broadcast(Feed.bestBlock(this.height, this.blockTimestamp));
+
+            console.log(`[${this.label}] New block ${this.height}`);
+        }
+
+        this.feeds.broadcast(Feed.imported(node));
+
+        console.log(`[${this.label}] ${node.name} imported ${node.height}, block time: ${node.blockTime / 1000}s, average: ${node.average / 1000}s | latency ${node.latency}`);
+    }
+}

packages/backend/src/Feed.ts

@@ -0,0 +1,136 @@
+import * as WebSocket from 'ws';
+import * as EventEmitter from 'events';
+import Node from './Node';
+import { timestamp, Maybe, FeedMessage, Types, idGenerator } from '@dotstats/common';
+
+const nextId = idGenerator<Types.FeedId>();
+const { Actions } = FeedMessage;
+
+export default class Feed {
+    public id: Types.FeedId;
+
+    public chain: Maybe<Types.ChainLabel> = null;
+    public readonly events = new EventEmitter();
+
+    private socket: WebSocket;
+    private messages: Array<FeedMessage.Message> = [];
+
+    constructor(socket: WebSocket) {
+        this.id = nextId();
+        this.socket = socket;
+
+        socket.on('message', (data) => this.handleCommand(data.toString()));
+        socket.on('error', () => this.disconnect());
+        socket.on('close', () => this.disconnect());
+    }
+
+    public static bestBlock(height: Types.BlockNumber, ts: Types.Timestamp): FeedMessage.Message {
+        return {
+            action: Actions.BestBlock,
+            payload: [height, ts]
+        };
+    }
+
+    public static addedNode(node: Node): FeedMessage.Message {
+        return {
+            action: Actions.AddedNode,
+            payload: [node.id, node.nodeDetails(), node.nodeStats(), node.blockDetails()]
+        };
+    }
+
+    public static removedNode(node: Node): FeedMessage.Message {
+        return {
+            action: Actions.RemovedNode,
+            payload: node.id
+        };
+    }
+
+    public static imported(node: Node): FeedMessage.Message {
+        return {
+            action: Actions.ImportedBlock,
+            payload: [node.id, node.blockDetails()]
+        };
+    }
+
+    public static stats(node: Node): FeedMessage.Message {
+        return {
+            action: Actions.NodeStats,
+            payload: [node.id, node.nodeStats()]
+        };
+    }
+
+    public static timeSync(): FeedMessage.Message {
+        return {
+            action: Actions.TimeSync,
+            payload: timestamp()
+        };
+    }
+
+    public static addedChain(label: Types.ChainLabel): FeedMessage.Message {
+        return {
+            action: Actions.AddedChain,
+            payload: label
+        };
+    }
+
+    public static removedChain(label: Types.ChainLabel): FeedMessage.Message {
+        return {
+            action: Actions.RemovedChain,
+            payload: label
+        }
+    }
+
+    public static subscribedTo(label: Types.ChainLabel): FeedMessage.Message {
+        return {
+            action: Actions.SubscribedTo,
+            payload: label,
+        }
+    }
+
+    public static unsubscribedFrom(label: Types.ChainLabel): FeedMessage.Message {
+        return {
+            action: Actions.UnsubscribedFrom,
+            payload: label,
+        }
+    }
+
+    public sendData(data: FeedMessage.Data) {
+        this.socket.send(data);
+    }
+
+    public sendMessage(message: FeedMessage.Message) {
+        const queue = this.messages.length === 0;
+
+        this.messages.push(message);
+
+        if (queue) {
+            process.nextTick(this.sendMessages);
+        }
+    }
+
+    private sendMessages = () => {
+        const data = FeedMessage.serialize(this.messages);
+        this.messages = [];
+        this.socket.send(data);
+    }
+
+    private handleCommand(cmd: string) {
+        if (cmd.startsWith('subscribe:')) {
+            if (this.chain) {
+                this.events.emit('unsubscribe', this.chain);
+                this.chain = null;
+            }
+
+            const label = cmd.substr(10) as Types.ChainLabel;
+
+            this.events.emit('subscribe', label);
+        }
+    }
+
+    private disconnect() {
+        this.socket.removeAllListeners();
+        this.socket.close();
+
+        this.events.emit('disconnect');
+    }
+}

packages/backend/src/FeedSet.ts

@@ -0,0 +1,55 @@
+import Feed from './Feed';
+import { FeedMessage } from '@dotstats/common';
+
+type DisconnectListener = () => void;
+
+export default class FeedSet {
+    private feeds = new Map<Feed, DisconnectListener>();
+    private messages: Array<FeedMessage.Message> = [];
+
+    public values(): IterableIterator<Feed> {
+        return this.feeds.keys();
+    }
+
+    public each(fn: (feed: Feed) => void) {
+        for (const feed of this.values()) {
+            fn(feed);
+        }
+    }
+
+    public add(feed: Feed) {
+        const listener = () => this.remove(feed);
+
+        this.feeds.set(feed, listener);
+
+        feed.events.once('disconnect', listener);
+    }
+
+    public remove(feed: Feed) {
+        const listener = this.feeds.get(feed);
+
+        if (!listener) {
+            return;
+        }
+
+        feed.events.removeListener('disconnect', listener);
+
+        this.feeds.delete(feed);
+    }
+
+    public broadcast(message: FeedMessage.Message) {
+        const queue = this.messages.length === 0;
+
+        this.messages.push(message);
+
+        if (queue) {
+            process.nextTick(this.sendMessages);
+        }
+    }
+
+    private sendMessages = () => {
+        const data = FeedMessage.serialize(this.messages);
+        this.messages = [];
+        this.each(feed => feed.sendData(data));
+    }
+}

packages/backend/src/Node.ts

@@ -8,12 +8,16 @@ const TIMEOUT = (1000 * 60 * 1) as Types.Milliseconds; // 1 minute
 
 const nextId = idGenerator<Types.NodeId>();
 
-export default class Node extends EventEmitter {
+export default class Node {
+    public readonly id: Types.NodeId;
+    public readonly name: Types.NodeName;
+    public readonly chain: Types.ChainLabel;
+    public readonly implementation: Types.NodeImplementation;
+    public readonly version: Types.NodeVersion;
+
+    public readonly events = new EventEmitter();
+
     public lastMessage: Types.Timestamp;
-    public id: Types.NodeId;
-    public name: Types.NodeName;
-    public implementation: Types.NodeImplementation;
-    public version: Types.NodeVersion;
     public config: string;
     public best = '' as Types.BlockHash;
     public height = 0 as Types.BlockNumber;
@@ -24,28 +28,26 @@ export default class Node extends EventEmitter {
     private peers = 0 as Types.PeerCount;
     private txcount = 0 as Types.TransactionCount;
 
-    private socket: WebSocket;
+    private readonly socket: WebSocket;
     private blockTimes: Array<number> = new Array(BLOCK_TIME_HISTORY);
     private lastBlockAt: Maybe<Date> = null;
 
     constructor(
         socket: WebSocket,
         name: Types.NodeName,
+        chain: Types.ChainLabel,
         config: string,
         implentation: Types.NodeImplementation,
         version: Types.NodeVersion,
     ) {
-        super();
-
-        this.lastMessage = timestamp();
         this.id = nextId();
-        this.socket = socket;
         this.name = name;
+        this.chain = chain;
         this.config = config;
         this.implementation = implentation;
         this.version = version;
-
-        console.log(`Listening to a new node: ${name}`);
+        this.lastMessage = timestamp();
+        this.socket = socket;
 
         socket.on('message', (data) => {
             const message = parseMessage(data);
@@ -94,9 +96,9 @@ export default class Node extends EventEmitter {
                 if (message && message.msg === "system.connected") {
                     cleanup();
 
-                    const { name, config, implementation, version } = message;
+                    const { name, chain, config, implementation, version } = message;
 
-                    resolve(new Node(socket, name, config, implementation, version));
+                    resolve(new Node(socket, name, chain, config, implementation, version));
                 }
             }
 
@@ -160,7 +162,7 @@ export default class Node extends EventEmitter {
         this.socket.removeAllListeners();
         this.socket.close();
 
-        this.emit('disconnect');
+        this.events.emit('disconnect');
     }
 
     private onSystemInterval(message: SystemInterval) {
@@ -170,7 +172,7 @@ export default class Node extends EventEmitter {
             this.peers = peers;
             this.txcount = txcount;
 
-            this.emit('stats');
+            this.events.emit('stats');
         }
     }
 
@@ -191,7 +193,7 @@ export default class Node extends EventEmitter {
             this.blockTimes[height % BLOCK_TIME_HISTORY] = blockTime;
             this.blockTime = blockTime;
 
-            this.emit('block');
+            this.events.emit('block');
         }
     }
 

packages/backend/src/aggregator.ts

@@ -1,96 +0,0 @@
-import * as EventEmitter from 'events';
-import Node from './node';
-import Feed from './feed';
-import { timestamp, Types, FeedMessage } from '@dotstats/common';
-
-export default class Aggregator extends EventEmitter {
-    private nodes = new Set<Node>();
-    private feeds = new Set<Feed>();
-    private messages: Array<FeedMessage.Message> = [];
-
-    public height = 0 as Types.BlockNumber;
-    public blockTimestamp = 0 as Types.Timestamp;
-
-    constructor() {
-        super();
-
-        setInterval(() => this.timeoutCheck(), 10000);
-    }
-
-    public addNode(node: Node) {
-        this.nodes.add(node);
-        this.broadcast(Feed.addedNode(node));
-
-        node.once('disconnect', () => {
-            node.removeAllListeners();
-
-            this.nodes.delete(node);
-            this.broadcast(Feed.removedNode(node));
-        });
-
-        node.on('block', () => this.updateBlock(node));
-        node.on('stats', () => this.broadcast(Feed.stats(node)));
-    }
-
-    public addFeed(feed: Feed) {
-        this.feeds.add(feed);
-
-        const messages = [Feed.timeSync(), Feed.bestBlock(this.height, this.blockTimestamp)];
-
-        for (const node of this.nodes.values()) {
-            messages.push(Feed.addedNode(node));
-        }
-
-        feed.sendMessages(messages);
-
-        feed.once('disconnect', () => {
-            this.feeds.delete(feed);
-        });
-    }
-
-    public nodeList(): IterableIterator<Node> {
-        return this.nodes.values();
-    }
-
-    private broadcast(message: FeedMessage.Message) {
-        const queue = this.messages.length === 0;
-
-        this.messages.push(message);
-
-        if (queue) {
-            process.nextTick(() => {
-                const data = FeedMessage.serialize(this.messages);
-                this.messages = [];
-
-                for (const feed of this.feeds.values()) {
-                    feed.sendData(data);
-                }
-            });
-        }
-    }
-
-    private timeoutCheck() {
-        const now = timestamp();
-
-        for (const node of this.nodes.values()) {
-            node.timeoutCheck(now);
-        }
-
-        this.broadcast(Feed.timeSync());
-    }
-
-    private updateBlock(node: Node) {
-        if (node.height > this.height) {
-            this.height = node.height;
-            this.blockTimestamp = node.blockTimestamp;
-
-            this.broadcast(Feed.bestBlock(this.height, this.blockTimestamp));
-
-            console.log(`New block ${this.height}`);
-        }
-
-        this.broadcast(Feed.imported(node));
-
-        console.log(`${node.name} imported ${node.height}, block time: ${node.blockTime / 1000}s, average: ${node.average / 1000}s | latency ${node.latency}`);
-    }
-}

packages/backend/src/feed.ts

@@ -1,80 +0,0 @@
-import * as WebSocket from 'ws';
-import * as EventEmitter from 'events';
-import Node from './node';
-import { timestamp, Opaque, FeedMessage, Types, idGenerator } from '@dotstats/common';
-
-const nextId = idGenerator<Types.FeedId>();
-const { Actions } = FeedMessage;
-
-export default class Feed extends EventEmitter {
-    public id: Types.FeedId;
-
-    private socket: WebSocket;
-
-    constructor(socket: WebSocket) {
-        super();
-
-        this.id = nextId();
-        this.socket = socket;
-
-        socket.on('error', () => this.disconnect());
-        socket.on('close', () => this.disconnect());
-    }
-
-    public static bestBlock(height: Types.BlockNumber, ts: Types.Timestamp): FeedMessage.Message {
-        return {
-            action: Actions.BestBlock,
-            payload: [height, ts]
-        };
-    }
-
-    public static addedNode(node: Node): FeedMessage.Message {
-        return {
-            action: Actions.AddedNode,
-            payload: [node.id, node.nodeDetails(), node.nodeStats(), node.blockDetails()]
-        };
-    }
-
-    public static removedNode(node: Node): FeedMessage.Message {
-        return {
-            action: Actions.RemovedNode,
-            payload: node.id
-        };
-    }
-
-    public static imported(node: Node): FeedMessage.Message {
-        return {
-            action: Actions.ImportedBlock,
-            payload: [node.id, node.blockDetails()]
-        };
-    }
-
-    public static stats(node: Node): FeedMessage.Message {
-        return {
-            action: Actions.NodeStats,
-            payload: [node.id, node.nodeStats()]
-        };
-    }
-
-    public static timeSync(): FeedMessage.Message {
-        return {
-            action: Actions.TimeSync,
-            payload: timestamp()
-        };
-    }
-
-    public sendData(data: FeedMessage.Data) {
-        this.socket.send(data);
-    }
-
-    public sendMessages(messages: Array<FeedMessage.Message>) {
-        this.socket.send(FeedMessage.serialize(messages))
-    }
-
-    private disconnect() {
-        this.socket.removeAllListeners();
-        this.socket.close();
-
-        this.emit('disconnect');
-    }
-}

packages/backend/src/index.ts

@@ -1,9 +1,9 @@
 import * as WebSocket from 'ws';
-import Node from './node';
-import Feed from './feed';
-import Aggregator from './aggregator';
+import Node from './Node';
+import Feed from './Feed';
+import Aggregator from './Aggregator';
 
-const aggregator = new Aggregator;
+const aggregator = new Aggregator();
 
 // WebSocket for Nodes feeding telemetry data to the server
 const incomingTelemetry = new WebSocket.Server({ port: 1024 });
@@ -16,7 +16,9 @@ console.log('Feed server listening on port 8080');
 
 incomingTelemetry.on('connection', async (socket: WebSocket) => {
     try {
-        aggregator.addNode(await Node.fromSocket(socket));
+        const node = await Node.fromSocket(socket);
+
+        aggregator.addNode(node);
     } catch (err) {
         console.error(err);
     }

packages/backend/src/message.ts

@@ -42,7 +42,7 @@ export interface BestBlock {
 interface SystemConnected {
     msg: 'system.connected',
     name: Types.NodeName,
-    chain: string,
+    chain: Types.ChainLabel,
     config: string,
     implementation: Types.NodeImplementation,
     version: Types.NodeVersion,