Merge branch '0.2.2'

lib/cube/emitter-http.js

@@ -0,0 +1,49 @@
+var util = require("util"),
+    http = require("http");
+
+module.exports = function(protocol, host, port) {
+  var emitter = {},
+      queue = [],
+      closing;
+
+  if (protocol != "http:") throw new Error("invalid HTTP protocol");
+
+  function send() {
+    var events = queue.splice(0, 500),
+        body = JSON.stringify(events);
+
+    http.request({
+      host: host,
+      port: port,
+      path: "/1.0/event",
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        "Content-Length": body.length
+      }
+    }, function(response) {
+      if (response.statusCode !== 200) return error(response.statusCode);
+      if (queue.length) setTimeout(send, 500);
+    }).on("error", function(e) {
+      error(e.message);
+    }).end(body);
+
+    function error(message) {
+      util.log("error: " + message);
+      queue.unshift.apply(queue, events);
+      setTimeout(send, 1000);
+    }
+  }
+
+  emitter.send = function(event) {
+    if (!closing && queue.push(event) === 1) setTimeout(send, 500);
+    return emitter;
+  };
+
+  emitter.close = function () {
+    if (queue.length) closing = 1;
+    return emitter;
+  };
+
+  return emitter;
+};

lib/cube/emitter.js

@@ -1,5 +1,6 @@
 var util = require("util"),
     url = require("url"),
+    http = require("./emitter-http"),
     udp = require("./emitter-udp"),
     ws = require("./emitter-ws");
 
@@ -9,6 +10,7 @@ module.exports = function(u) {
   switch (u.protocol) {
     case "udp:": emitter = udp; break;
     case "ws:": case "wss:": emitter = ws; break;
+    case "http:": emitter = http; break;
   }
   return emitter(u.protocol, u.hostname, u.port);
 };

lib/cube/event.js

@@ -106,7 +106,8 @@ exports.putter = function(db) {
         var tierTimes = timesToInvalidateByTier[tier],
             tierTime = tiers[tier].floor(time),
             i = bisect(tierTimes, tierTime);
-        if (tierTimes[i] > tierTime) tierTimes.splice(i, 0, tierTime);
+        if (i >= tierTimes.length) tierTimes.push(tierTime);
+        else if (tierTimes[i] > tierTime) tierTimes.splice(i, 0, tierTime);
       }
     } else {
       timesToInvalidateByTier = timesToInvalidateByTierByType[type] = {};

lib/cube/server.js

@@ -131,9 +131,9 @@ module.exports = function(options) {
     // If this request wasn't matched, see if there's a static file to serve.
     request.on("end", function() {
       file.serve(request, response, function(error) {
-        if (error && error.status == 404) {
-          response.writeHead(404, {"Content-Type": "text/plain"});
-          response.end("404 Not Found");
+        if (error) {
+          response.writeHead(error.status, {"Content-Type": "text/plain"});
+          response.end(error.status + "");
         }
       });
     });

package.json

@@ -1,14 +1,14 @@
 {
   "name": "cube",
-  "version": "0.2.1",
+  "version": "0.2.2",
   "description": "A system for analyzing time series data using MongoDB and Node.",
   "keywords": ["time series"],
   "homepage": "http://square.github.com/cube/",
   "author": {"name": "Mike Bostock", "url": "http://bost.ocks.org/mike"},
   "repository": {"type": "git", "url": "http://github.com/square/cube.git"},
   "main": "./lib/cube",
   "dependencies": {
-    "mongodb": "0.9.9-8",
+    "mongodb": "1.0.1",
     "node-static": "0.5.9",
     "pegjs": "0.6.2",
     "vows": "0.5.11",