redirector: handle closed connections, refactor into RpcConnSet util

Author: guymguym

src/rpc/rpc_conn_set.js

@@ -0,0 +1,52 @@
+/* Copyright (C) 2016 NooBaa */
+'use strict';
+
+const dbg = require('../util/debug_module')(__filename);
+
+const CLOSE_LISTENER_SYMBOL = Symbol('CLOSE_LISTENER_SYMBOL');
+
+class RpcConnSet {
+
+    constructor(name) {
+        this.name = name;
+        this.set = new Set();
+    }
+
+    add(conn) {
+        if (conn.is_closed()) {
+            dbg.warn(this.name, 'ignore closed connection', conn.url.href);
+            return;
+        }
+        if (this.set.has(conn)) {
+            dbg.log0(this.name, 'already registered', conn.url.href);
+            return;
+        }
+        dbg.log0(this.name, 'adding connection', conn.url.href);
+        this.set.add(conn);
+        const close_listener = () => this.remove(conn);
+        conn[CLOSE_LISTENER_SYMBOL] = close_listener;
+        conn.once('close', close_listener);
+    }
+
+    remove(conn) {
+        dbg.warn(this.name, 'removing connection', conn.url.href);
+        const close_listener = conn[CLOSE_LISTENER_SYMBOL];
+        delete conn[CLOSE_LISTENER_SYMBOL];
+        conn.removeListener('close', close_listener);
+        this.set.delete(conn);
+    }
+
+    cleanup() {
+        for (const conn of this.set) {
+            if (conn.is_closed()) this.remove(conn);
+        }
+    }
+
+    list() {
+        this.cleanup();
+        return Array.from(this.set);
+    }
+
+}
+
+module.exports = RpcConnSet;

src/server/system_services/redirector.js

@@ -1,87 +1,52 @@
 /* Copyright (C) 2016 NooBaa */
-/**
- *
- * REDIRECTOR
- *
- */
 'use strict';
 
-const _ = require('lodash');
 const P = require('../../util/promise');
 const dbg = require('../../util/debug_module')(__filename);
 const server_rpc = require('../server_rpc');
+const RpcConnSet = require('../../rpc/rpc_conn_set');
 
-// dbg.set_level(5);
-
-const cluster_connections = new Set();
-const alerts_connections = new Set();
+const cluster_conn_set = new RpcConnSet('redirector cluster_conn_set');
+const alerts_conn_set = new RpcConnSet('redirector alerts_conn_set');
 
 function register_to_cluster(req) {
-    var conn = req.connection;
-    if (!cluster_connections.has(conn)) {
-        dbg.log0('register_to_cluster', conn.url.href);
-        cluster_connections.add(conn);
-        conn.on('close', function() {
-            cluster_connections.delete(conn);
-        });
-    }
+    cluster_conn_set.add(req.connection);
 }
 
 function publish_to_cluster(req) {
-    var api_name = req.rpc_params.method_api.slice(0, -4); // remove _api suffix
-    var method = req.rpc_params.method_name;
-    var addresses = ['fcall://fcall']; // also call on myself
-    cluster_connections.forEach(function(conn) {
-        addresses.push(conn.url.href);
-    });
-    addresses = _.uniq(addresses);
-    dbg.log0('publish_to_cluster:', addresses);
-    return P.map(addresses, function(address) {
-            return server_rpc.client[api_name][method](req.rpc_params.request_params, {
-                address: address,
+    const api_name = req.rpc_params.method_api.slice(0, -4); // remove _api suffix
+    const method = req.rpc_params.method_name;
+    const connections = cluster_conn_set.list();
+    dbg.log0('publish_to_cluster:', connections.length);
+    return P.map(connections,
+            conn => server_rpc.client[api_name][method](req.rpc_params.request_params, {
+                connection: conn,
                 auth_token: req.auth_token,
-            });
-        })
-        .then(function(res) {
-            return {
-                redirect_reply: {
-                    aggregated: res,
-                }
-            };
-        });
+            })
+        )
+        .then(res => ({
+            redirect_reply: {
+                aggregated: res,
+            }
+        }));
 }
 
 function register_for_alerts(req) {
-    var conn = req.connection;
-    if (!alerts_connections.has(conn)) {
-        dbg.log0('register_for_alerts', conn.url.href);
-        alerts_connections.add(conn);
-        conn.on('close', function() {
-            alerts_connections.delete(conn);
-        });
-    }
+    alerts_conn_set.add(req.connection);
 }
 
 function unregister_from_alerts(req) {
-    var conn = req.connection;
-    if (!alerts_connections.has(conn)) {
-        return;
-    }
-    alerts_connections.delete(conn);
+    alerts_conn_set.remove(req.connection);
 }
 
 function publish_alerts(req) {
-    var connections = [];
-    alerts_connections.forEach(function(conn) {
-        connections.push(conn);
-    });
-    connections = _.uniq(connections);
-    dbg.log3('publish_alerts:', req.rpc_params.request_params, connections);
-    return P.map(connections, function(conn) {
-            return server_rpc.client.frontend_notifications.alert(req.rpc_params.request_params, {
+    const connections = alerts_conn_set.list();
+    dbg.log3('publish_alerts:', req.rpc_params.request_params, connections.length);
+    return P.map(connections, conn =>
+            server_rpc.client.frontend_notifications.alert(req.rpc_params.request_params, {
                 connection: conn,
-            });
-        })
+            })
+        )
         .then(() => {
             dbg.log3('published');
         });