Merge pull request #81 from pseudomuto/separate_route_storage

Separate storage mechanism for routes

Author: minrk

lib/configproxy.js

@@ -17,7 +17,7 @@ var http = require('http'),
     util = require('util'),
     URL = require('url'),
     querystring = require('querystring'),
-    trie = require('./trie.js');
+    store = require('./store.js');
 
 function bound (that, method) {
     // bind a method, to ensure `this=that` when it is called
@@ -100,10 +100,10 @@ function parse_host (req) {
 function ConfigurableProxy (options) {
     var that = this;
     this.options = options || {};
-    this.trie = new trie.URLTrie();
+
+    this._routes = store.MemoryStore();
     this.auth_token = this.options.auth_token;
     this.includePrefix = options.includePrefix === undefined ? true : options.includePrefix;
-    this.routes = {};
     this.host_routing = this.options.host_routing;
     this.error_target = options.error_target;
     if (this.error_target && this.error_target.slice(-1) !== '/') {
@@ -169,7 +169,7 @@ function ConfigurableProxy (options) {
     }
 
     // proxy requests separately
-    var proxy_callback = log_errors(that.handle_proxy_web);
+    var proxy_callback = log_errors(this.handle_proxy_web);
     if ( this.options.ssl ) {
         this.proxy_server = https.createServer(this.options.ssl, proxy_callback);
     } else {
@@ -185,23 +185,33 @@ function ConfigurableProxy (options) {
 
 util.inherits(ConfigurableProxy, EventEmitter);
 
-ConfigurableProxy.prototype.add_route = function (path, data) {
+ConfigurableProxy.prototype.add_route = function (path, data, cb) {
     // add a route to the routing table
-    path = trie.trim_prefix(path);
+    path = this._routes.cleanPath(path);
     if (this.host_routing && path !== '/') {
         data.host = path.split('/')[1];
     }
-    this.routes[path] = data;
-    this.trie.add(path, data);
-    this.update_last_activity(path);
+
+    var that = this;
+
+    this._routes.add(path, data, function () {
+      that.update_last_activity(path, function () {
+        if (typeof(cb) === "function") {
+          cb();
+        }
+      });
+    });
 };
 
-ConfigurableProxy.prototype.remove_route = function (path) {
+ConfigurableProxy.prototype.remove_route = function (path, cb) {
     // remove a route from the routing table
-    if (this.routes[path] !== undefined) {
-        delete this.routes[path];
-        this.trie.remove(path);
-    }
+    var routes = this._routes;
+
+    routes.hasRoute(path, function (result) {
+        if (result) {
+            routes.remove(path, cb);
+        }
+    });
 };
 
 ConfigurableProxy.prototype.get_routes = function (req, res) {
@@ -211,7 +221,7 @@ ConfigurableProxy.prototype.get_routes = function (req, res) {
     var inactive_since = null;
     if (parsed.query) {
         var query = querystring.parse(parsed.query);
-        
+
         if (query.inactive_since !== undefined) {
             var timestamp = Date.parse(query.inactive_since);
             if (isFinite(timestamp)) {
@@ -223,20 +233,24 @@ ConfigurableProxy.prototype.get_routes = function (req, res) {
         }
     }
     res.writeHead(200, { 'Content-Type': 'application/json' });
-    var routes = {};
-    if (inactive_since) {
-        Object.keys(this.routes).map(function (path) {
-            var route = that.routes[path];
-            if (route.last_activity < inactive_since) {
-                routes[path] = route;
-            }
-        });
-    } else {
-        routes = this.routes;
-    }
-    res.write(JSON.stringify(routes));
-    res.end();
-    this.statsd.increment('api.route.get', 1);
+
+    this._routes.getAll(function (routes) {
+        var results = {};
+
+        if (inactive_since) {
+            Object.keys(routes).forEach(function (path) {
+                if (routes[path].last_activity < inactive_since) {
+                    results[path] = routes[path];
+                }
+            });
+        } else {
+            results = routes;
+        }
+
+        res.write(JSON.stringify(results));
+        res.end();
+        that.statsd.increment('api.route.get', 1);
+    });
 };
 
 ConfigurableProxy.prototype.post_routes = function (req, res, path, data) {
@@ -250,47 +264,70 @@ ConfigurableProxy.prototype.post_routes = function (req, res, path, data) {
         return;
     }
 
-    this.add_route(path, data);
-    res.writeHead(201);
-    res.end();
-    this.statsd.increment('api.route.add', 1);
+    var that = this;
+    this.add_route(path, data, function () {
+        res.writeHead(201);
+        res.end();
+        that.statsd.increment('api.route.add', 1);
+    });
 };
 
 ConfigurableProxy.prototype.delete_routes = function (req, res, path) {
     // DELETE removes an existing route
     log.debug('DELETE', path);
-    if (this.routes[path] === undefined) {
-        res.writeHead(404);
-    } else {
-        this.remove_route(path);
-        res.writeHead(204);
-    }
-    res.end();
-    this.statsd.increment('api.route.delete', 1);
+
+    var that = this;
+    this._routes.hasRoute(path, function (result) {
+        if (result) {
+            that.remove_route(path, function () {
+                res.writeHead(204);
+                res.end();
+                that.statsd.increment('api.route.delete', 1);
+            });
+        } else {
+            res.writeHead(404);
+            res.end();
+            that.statsd.increment('api.route.delete', 1);
+        }
+    });
 };
 
-ConfigurableProxy.prototype.target_for_req = function (req) {
+ConfigurableProxy.prototype.target_for_req = function (req, cb) {
     var timer = this.statsd.createTimer('find_target_for_req');
     // return proxy target for a given url path
     var base_path = (this.host_routing) ? '/' + parse_host(req) : '';
-    var route = this.trie.get(trie.string_to_path(base_path + decodeURIComponent(req.url)));
-    timer.stop();
-    if (route) {
-        return {
-            prefix: route.prefix,
-            target: route.data.target,
-        };
-    }
+
+    this._routes.getTarget(base_path + decodeURIComponent(req.url), function (route) {
+        timer.stop();
+        if (route) {
+            cb({
+                prefix: route.prefix,
+                target: route.data.target
+            });
+            return;
+        }
+
+        cb(null);
+    });
 };
 
-ConfigurableProxy.prototype.update_last_activity = function (prefix) {
+ConfigurableProxy.prototype.update_last_activity = function (prefix, cb) {
     var timer = this.statsd.createTimer('last_activity_updating');
-    // note last activity in routing table
-    if (this.routes[prefix] !== undefined) {
-        // route may have been deleted with open connections
-        this.routes[prefix].last_activity = new Date();
-    }
-    timer.stop();
+    var routes = this._routes;
+
+    routes.hasRoute(prefix, function (result) {
+        cb = cb || function() {};
+
+        if (result) {
+            routes.update(prefix, { "last_activity": new Date() }, function () {
+                timer.stop();
+                cb();
+            });
+        } else {
+            timer.stop();
+            cb();
+        }
+    });
 };
 
 ConfigurableProxy.prototype._handle_proxy_error_default = function (code, kind, req, res) {
@@ -363,48 +400,47 @@ ConfigurableProxy.prototype.handle_proxy_error = function (code, kind, req, res)
 ConfigurableProxy.prototype.handle_proxy = function (kind, req, res) {
     // proxy any request
     var that = this;
+    var args = Array.prototype.slice.call(arguments, 1);
+
     // get the proxy target
-    var match = this.target_for_req(req);
-    if (!match) {
-        this.handle_proxy_error(404, kind, req, res);
-        return;
-    }
-    this.emit("proxy_request", req, res);
-    var prefix = match.prefix;
-    var target = match.target;
-    log.debug("PROXY", kind.toUpperCase(), req.url, "to", target);
-    if (!this.includePrefix) {
-        req.url = req.url.slice(prefix.length);
-    }
+    this.target_for_req(req, function (match) {
+        if (!match) {
+            that.handle_proxy_error(404, kind, req, res);
+            return;
+        }
 
-    // pop method off the front
-    var args = arguments_array(arguments);
-    args.shift();
+        that.emit("proxy_request", req, res);
+        var prefix = match.prefix;
+        var target = match.target;
+        log.debug("PROXY", kind.toUpperCase(), req.url, "to", target);
+        if (!that.includePrefix) {
+            req.url = req.url.slice(prefix.length);
+        }
 
-    // add config argument
-    args.push({
-        target: target
-    });
+        // add config argument
+        args.push({ target: target });
 
-    // add error handling
-    args.push(function (e) {
-        log.error("Proxy error: ", e);
-        that.handle_proxy_error(503, kind, req, res);
-    });
+        // add error handling
+        args.push(function (e) {
+            log.error("Proxy error: ", e);
+            that.handle_proxy_error(503, kind, req, res);
+        });
 
-    // update last activity timestamp in routing table
-    this.update_last_activity(prefix);
+        // update timestamp on any reply data as well (this includes websocket data)
+        req.on('data', function () {
+            that.update_last_activity(prefix);
+        });
 
-    // update timestamp on any reply data as well (this includes websocket data)
-    req.on('data', function () {
-        that.update_last_activity(prefix);
-    });
-    res.on('data', function () {
-        that.update_last_activity(prefix);
-    });
+        res.on('data', function () {
+            that.update_last_activity(prefix);
+        });
 
-    // dispatch the actual method
-    this.proxy[kind].apply(this.proxy, args);
+        // update last activity timestamp in routing table
+        that.update_last_activity(prefix, function () {
+            // dispatch the actual method
+            that.proxy[kind].apply(that.proxy, args);
+        });
+    });
 };
 
 ConfigurableProxy.prototype.handle_proxy_ws = function (req, res, head) {

lib/store.js

@@ -0,0 +1,84 @@
+var trie = require("./trie.js");
+
+var NotImplemented = function (name) {
+  return {
+    name: "NotImplementedException",
+    message: "method '" + name + "' not implemented"
+  };
+};
+
+var BaseStore = Object.create(Object.prototype, {
+  // "abstract" methods
+  get:       { value: function () { throw NotImplemented("get"); } },
+  getTarget: { value: function () { throw NotImplemented("getTarget"); } },
+  getAll:    { value: function () { throw NotImplemented("getAll"); } },
+  add:       { value: function () { throw NotImplemented("add"); } },
+  update:    { value: function () { throw NotImplemented("update"); } },
+  remove:    { value: function () { throw NotImplemented("remove"); } },
+  hasRoute:  { value: function () { throw NotImplemented("hasRoute"); } },
+
+  cleanPath: {
+    value: function (path) {
+      return trie.trim_prefix(path);
+    }
+  },
+
+  notify: {
+    value: function (cb) {
+      if (typeof(cb) === "function") {
+        var args = Array.prototype.slice.call(arguments, 1);
+        cb.apply(this, args);
+      }
+    }
+  }
+});
+
+function MemoryStore () {
+  var routes = {};
+  var urls   = new trie.URLTrie();
+
+  return Object.create(BaseStore, {
+    get: {
+      value: function (path, cb) {
+        this.notify(cb, routes[path]);
+      }
+    },
+    getTarget: {
+      value: function (path, cb) {
+        this.notify(cb, urls.get(path));
+      }
+    },
+    getAll: {
+      value: function (cb) {
+        this.notify(cb, routes);
+      }
+    },
+    add: {
+      value: function (path, data, cb) {
+        routes[path] = data;
+        urls.add(path, data);
+        this.notify(cb);
+      }
+    },
+    update: {
+      value: function (path, data, cb) {
+        Object.assign(routes[path], data);
+        this.notify(cb);
+      }
+    },
+    remove: {
+      value: function (path, cb) {
+        delete routes[path];
+        urls.remove(path);
+        this.notify(cb);
+      }
+    },
+    hasRoute: {
+      value: function (path, cb) {
+        this.notify(cb, routes.hasOwnProperty(path));
+      }
+    }
+  });
+}
+
+exports.MemoryStore = MemoryStore;