introduced consumers polymorphism

config.yml

@@ -2,6 +2,7 @@
 
 connectors:
   - file: connectorRIS
+    name: ris
     params:
       url: wss://ris-live.ripe.net/v1/ws/
       moreSpecific: true

connectorFactory.js

@@ -3,33 +3,65 @@ import env from "./env";
 export default class ConnectorFactory {
 
     constructor() {
-        this.disconnected = [];
+        this.connectors = {};
-        this.connected = [];
-
     }
 
+    getConnector = (name) => {
+        return this.connectors[name];
+    };
+
+    getConnectors = () => {
+        return Object.keys(this.connectors).map(name => this.connectors[name]);
+    };
+
     loadConnectors = () => {
-        if (this.disconnected.length === 0) {
+        const connectors = Object.keys(this.connectors);
-            this.disconnected = config.reports.map(connector => new connector.class(connector.params, env));
+        if (connectors.length === 0) {
+
+            for (let connector of env.config.connectors) {
+                this.connectors[connector.name] = new connector.class(connector.params, env);
+            }
         }
     };
 
     connectConnectors = () =>
-        Promise.all(this.disconnected.map(connector => {
+        new Promise((resolve, reject) => {
-            connector.connect()
+            const connectors = this.getConnectors();
-                .then(() => {
-                    this.connected.push();
-                });
-        }));
 
-    subscribeConnectors = (params) =>
+            if (connectors.length === 0) {
+                reject(new Error("No connections available"));
+
+            } else {
+                resolve(Promise.all(connectors
+                    .map(connector =>
+                        new Promise((resolve, reject) => {
+                            connector.connect()
+                                .then(() => {
+                                    connector.connected = true;
+                                    resolve(true);
+                                })
+                                .catch((error) => {
+                                    env.logger.log({
+                                        level: 'error',
+                                        message: error
+                                    });
+                                    resolve(false);
+                                })
+                        }))));
+            }
+        });
+
+    subscribeConnectors = (params, callback) =>
         new Promise((resolve, reject) => {
 
-            if (this.connectors.length === 0) {
+            const connectors = this.getConnectors();
-                reject(new Error("No connectors loaded"));
-            } else {
 
-                resolve(Promise.all(this.connectors.map(connector => connector.subscribe(params))));
+            if (connectors.length === 0) {
+                reject(new Error("No connections available"));
+            } else {
+                resolve(Promise.all(connectors.map(connector => {
+                    connector.subscribe(params);
+                })));
             }
 
         });

connectors/connector.js

@@ -1,5 +1,4 @@
 
-
 export default class Connector {
 
     constructor(params, env){
@@ -16,7 +15,7 @@ export default class Connector {
         new Promise((resolve, reject) => reject(new Error('The method connect has to be implemented')));
 
 
-    close = () => {
+    error = () => {
         this.logger.log({
             level: 'info',
             message: 'Web socket disconnected'
@@ -27,7 +26,15 @@ export default class Connector {
         throw new Error('The method subscribe has to be implemented');
     };
 
-    message = (message) => this.messageCallback(message);
+    message = (message) => {
+        if (this.messageCallback)
+        this.messageCallback(message);
+    };
+
+    connected = (message) => {
+        if (this.connectCallback)
+            this.connectCallback(message);
+    };
 
     transform = (message) => {
         throw new Error('The method transform has to be implemented');
@@ -41,7 +48,8 @@ export default class Connector {
         this.messageCallback = callback;
     };
 
-    onClose = (callback) => {
+    onError = (callback) => {
         this.closeCallback = callback;
     };
+
 }

connectors/connectorRIS.js

@@ -1,32 +1,44 @@
 import WebSocket from "ws";
+import Connector from "./connector";
 
 export default class ConnectorRIS extends Connector{
 
     constructor(params, env) {
         super(params, env);
+        this.ws = null;
     }
 
-    connect = () => {
+    connect = () =>
+        new Promise((resolve, reject) => {
+            try {
 
-        // const ws = new WebSocket(this.params.url);
+                this.ws = new WebSocket(this.params.url);
-        //
+                this.ws.on('message', this.message);
-        // ws.on('message', this.message);
+                this.ws.on('close', this.error);
-        //
+                this.ws.on('open', () => {
-        // ws.on('open', () => {
+                    resolve(true);
-        //
+                    this.connected();
-        // });
+                });
-        //
-        // ws.on('close', this.close);
 
-        new Promise((resolve, reject) => reject(new Error('The method connect has to be implemented')));
+            } catch(error) {
-    };
+                resolve(false);
+            }
+        });
+
+
+    subscribe = (input) =>
+        new Promise((resolve, reject) => {
+            try {
+                this.ws.send(JSON.stringify({
+                    type: "ris_subscribe",
+                    data: this.params
+                }));
+                resolve(true);
+            } catch(error) {
+                resolve(false);
+            }
+        });
 
-    subscribe = (input) => {
-        ws.send(JSON.stringify({
-            type: "ris_subscribe",
-            data: this.params
-        }));
-    };
 
     transform = (message) => {
         message = message.data;

connectors/connectorTest.js

@@ -0,0 +1,67 @@
+import WebSocket from "ws";
+
+export default class ConnectorRIS extends Connector{
+
+    constructor(params, env) {
+        super(params, env);
+    }
+
+    connect = () => {
+
+        // const ws = new WebSocket(this.params.url);
+        //
+        // ws.on('message', this.message);
+        //
+        // ws.on('open', () => {
+        //
+        // });
+        //
+        // ws.on('close', this.close);
+
+        new Promise((resolve, reject) => reject(new Error('The method connect has to be implemented')));
+    };
+
+    subscribe = (input) => {
+        ws.send(JSON.stringify({
+            type: "ris_subscribe",
+            data: this.params
+        }));
+    };
+
+    transform = (message) => {
+        message = message.data;
+        const components = [];
+        const announcements = message["announcements"] || [];
+        const withdrawals = message["withdrawals"] || [];
+        const peer = message["peer"];
+        const path = message["path"];
+
+        for (let announcement of announcements){
+            const nextHop = announcement["next_hop"];
+            const prefixes = announcement["prefixes"] || [];
+
+            for (let prefix of prefixes){
+                components.push({
+                    type: "announcement",
+                    prefix,
+                    peer,
+                    path,
+                    originAs: path[path.length - 1],
+                    nextHop
+                })
+            }
+        }
+
+        for (let prefix of withdrawals){
+            components.push({
+                type: "withdrawal",
+                prefix,
+                peer
+
+            })
+        }
+
+        return components;
+    };
+
+}

consumer.js

@@ -21,6 +21,8 @@ export default class Consumer {
     };
 
     handleUpdate = (data) => {
+        // console.log(data);
+        return;
         const messages = this.transform(data);
         for (let monitor of monitors) {
 
@@ -73,7 +75,7 @@ export default class Consumer {
         }
 
         return components;
-    }
+    };
 
 }
 

env.js

@@ -53,7 +53,10 @@ const logger = winston.createLogger({
     level: 'info',
     transports: [
         transportError,
-        transportReports
+        transportReports,
+        new winston.transports.Console({
+            format: winston.format.simple()
+        })
     ]
 });
 
@@ -85,7 +88,8 @@ config.connectors = (config.connectors || [])
 
         return {
             class: require("./connectors/" + item.file).default,
-            params: item.params
+            params: item.params,
+            name: item.name
         };
 
     });
@@ -96,8 +100,8 @@ vector.config = config;
 vector.logger = logger;
 vector.input = input;
 vector.pubSub = pubSub;
-vector.monitors = config.monitors.map(monitor => new monitor.class(monitor.name, monitor.channel, vector));
+// vector.monitors = config.monitors.map(monitor => new monitor.class(monitor.name, monitor.channel, vector));
-vector.reports = config.reports.map(report => new report.class(report.channels, vector));
+// vector.reports = config.reports.map(report => new report.class(report.channels, vector));
 
 
 module.exports = vector;

index.js

@@ -1,65 +1,42 @@
-import { config, logger } from "./env";
+import { config, logger, input } from "./env";
 import cluster from "cluster";
 import WebSocket from "ws";
 import sleep from "sleep";
 import Consumer from "./consumer";
-import Connector from "./connector";
+import ConnectorFactory from "./connectorFactory";
 
 if (cluster.isMaster) {
 
     const worker = cluster.fork();
 
 
-    if (config.testMode){
+    const connectorFactory = new ConnectorFactory();
-        // const update = {
-        //     data: {
-        //         withdrawals: ["124.40.52.0/22"],
-        //         peer: "124.0.0.2"
-        //     },
-        //     type: "ris_message"
-        // };
 
-        const update = {
+    connectorFactory.loadConnectors();
-            data: {
+    connectorFactory.connectConnectors()
-                announcements: [{
+        .then(() => connectorFactory.subscribeConnectors(input))
-                    prefixes: ["124.40.52.0/22"],
+        .then(() => {
-                    next_hop: "124.0.0.2"
-                }],
-                peer: "124.0.0.2",
-                path: "1,2,3,2914".split(",")
-            },
-            type: "ris_message"
-        };
 
-        const message = JSON.stringify(update);
+            for (const connector of connectorFactory.getConnectors()) {
+                connector.onMessage((message) => {
+                    worker.send(message);
+                });
+                connector.onError(error => {
+                    logger.log({
+                        level: 'error',
+                        message: error
+                    });
+                });
 
-        while (true){
+            }
-            worker.send(message);
+        })
-            sleep.sleep(1);
+        .catch(error => {
-        }
-
-    } else {
-
-        const ws = new WebSocket(config.websocketDataService);
-
-        ws.on('message', (message) => {
-            worker.send(message);
-        });
-
-        ws.on('open', () => {
-            ws.send(JSON.stringify({
-                type: "ris_subscribe",
-                data: config.wsParams
-            }));
-        });
-
-        ws.on('close', function close() {
             logger.log({
-                level: 'info',
+                level: 'error',
-                message: 'Web socket disconnected'
+                message: error
             });
         });
-    }
+
 
 } else {
     new Consumer();