Better detection of socket closure.

lib/cube/event.js

@@ -110,6 +110,7 @@ exports.getter = function(db) {
     if (isNaN(stop)) return callback({error: "invalid stop"}), -1;
 
     // Parse the expression.
+    var expression;
     try {
       expression = parser.parse(request.expression);
     } catch (error) {
@@ -170,36 +171,29 @@ exports.getter = function(db) {
       else {
         streams = streamsBySource[expression.source] = {time: stop, waiting: [], active: [callback]};
         (function poll() {
-          query(function callback(event) {
+          query(function(event) {
 
-            // If there's an event…
+            // If there's an event, send it to all active, open clients.
             if (event) {
-              var closed = false;
-
-              // Send the event to all active, open clients.
               streams.active.forEach(function(callback) {
                 if (!callback.closed) callback(event);
-                else closed = true;
               });
-
-              // Remove any closed callbacks.
-              // Removal is rare, so we don't want to filter every time.
-              if (closed) streams.active = streams.active.filter(open);
-
-              // If no clients remain, then it's safe to close the callback.
-              // The query function will then terminate the underlying cursor.
-              if (!streams.active.length && !streams.waiting.length) {
-                callback.closed = true;
-                delete streamsBySource[expression.source];
-              }
             }
 
             // Otherwise, we've reached the end of a poll, and it's time to
             // merge the waiting callbacks into the active callbacks. Advance
             // the time range, and set a timeout for the next poll.
             else {
-              streams.active = streams.active.concat(streams.waiting);
+              streams.active = streams.active.concat(streams.waiting).filter(open);
               streams.waiting = [];
+
+              // If no clients remain, then it's safe to delete the shared
+              // stream, and we'll no longer be responsible for polling.
+              if (!streams.active.length) {
+                delete streamsBySource[expression.source];
+                return;
+              }
+
               filter.t.$gte = streams.time;
               filter.t.$lt = streams.time = new Date(Date.now() - delay);
               setTimeout(poll, streamInterval);

lib/cube/server.js

@@ -73,31 +73,15 @@ module.exports = function(options) {
       if ((e = endpoints.ws[i]).match(request.url)) {
 
         function callback(response) {
-          if (connection.socket.writable) {
-            connection.sendUTF(JSON.stringify(response));
-          }
+          connection.sendUTF(JSON.stringify(response));
         }
 
         callback.id = ++id;
 
-        // Listen for close events.
-        if (e.dispatch.close) {
-          connection.on("close", function() {
-            interval = clearInterval(interval);
-            e.dispatch.close(callback);
-          });
-
-          // Unfortunately, it looks like there is a bug in websocket-server (or
-          // somewhere else) where close events are not emitted if the socket is
-          // closed very shortly after it is opened. So we do an additional
-          // check using an interval to verify that the socket is still open.
-          var interval = setInterval(function() {
-            if (!connection.socket.writable) {
-              interval = clearInterval(interval);
-              connection.close();
-            }
-          }, 5000);
-        }
+        // Listen for socket disconnect.
+        if (e.dispatch.close) connection.socket.on("end", function() {
+          e.dispatch.close(callback);
+        });
 
         connection.on("message", function(message) {
           e.dispatch(JSON.parse(message.utf8Data || message), callback);