#73 Adopt Stream API

README.md

@@ -207,12 +207,13 @@ to a single process.
 
 #### Socket
 
-A representation of a client. _Inherits from EventEmitter_.
+A representation of a client. _Inherits from EventEmitter_. _Implements Stream_.
 
 ##### Events
 
 - `close`
     - Fired when the client is disconnected.
+    - `error` is fired prior to closing if the reason is other than server/transport/forced close
     - **Arguments**
       - `String`: reason for closing
       - `Object`: description object (optional)
@@ -260,6 +261,29 @@ A representation of a client. _Inherits from EventEmitter_.
 - `close`
     - Disconnects the client
     - **Returns** `Socket` for chaining
+- `pipe`
+    - Pipes incoming data to a Writable Stream.
+    - **Parameters**
+      - `WritableStream`: destination
+      - `Object`: optional, does not close destination if options.end is false
+    - **Returns** `Socket` for chaining
+
+##### Stream interface - see [Node.js docs](http://nodejs.org/api/stream.html#stream_event_drain)
+
+- Readable Stream
+    - supported:
+      - events: `data`, `end`, `error`, `close`
+      - properties: `readable`
+      - methods: `destroy`
+    - not supported:
+      - methods: `setEncoding`, `pause`, `resume`
+- Writable Stream
+    - supported:
+      - events: `error`, `close`
+      - properties: `writable`
+      - methods: `write`, `end`, `destroy`, `destroySoon`
+    - not supported:
+      - events: `drain`, `pipe`
 
 ### Client
 

lib/socket.js

@@ -22,9 +22,14 @@ function Socket (id, server, transport) {
   this.server = server;
   this.upgraded = false;
   this.readyState = 'opening';
+  this.readable = true;
+  this.writable = true;
   this.writeBuffer = [];
   this.packetsFn = [];
   this.sentCallbackFn = [];
+  this.on('error', function(){
+    debug('prevented error throwing');
+  });
 
   this.setTransport(transport);
   this.onOpen();
@@ -107,6 +112,26 @@ Socket.prototype.onPacket = function (packet) {
   }
 };
 
+
+/**
+ * Pipes output from this Socket to given destination WritableStream.
+ *
+ * @param {WritableStream} destination
+ * @param {Object} optional, does not close destination if options.end is false
+ * @return {Socket} for chaining
+ * @api public
+ */
+
+Socket.prototype.pipe = function (destination, options) {
+  if (destination.writable) {
+    this.on('data', destination.write.bind(destination));
+    if (!options || options.end !== false) {
+      this.on('end', destination.end.bind(destination));
+    }
+  }
+  return this;
+}
+
 /**
  * Called upon transport error.
  *
@@ -226,11 +251,19 @@ Socket.prototype.clearTransport = function () {
 
 Socket.prototype.onClose = function (reason, description) {
   if ('closed' != this.readyState) {
+    if (reason != 'server close' ||
+        reason != 'transport close' ||
+        reason != 'forced close') {
+      this.emit('error', description);
+    }
     this.packetsFn = [];
     this.sentCallbackFn = [];
     this.clearTransport();
     this.readyState = 'closed';
     this.emit('close', reason, description);
+    this.emit('end');
+    this.readable = false;
+    this.writable = false;
   }
 };
 
@@ -347,13 +380,32 @@ Socket.prototype.getAvailableUpgrades = function () {
   return availableUpgrades;
 };
 
+
+/**
+ * Writes any queued data and given data before closing the socket.
+ *
+ * @param {String} optional, data
+ * @return {Socket} for chaining
+ * @api public
+ */
+Socket.prototype.end = function (data) {
+  if (data) {
+    this.send(data);
+  } else {
+    this.flush();
+  }
+  this.close();
+  return this;
+}
+
 /**
  * Closes the socket and underlying transport.
  *
  * @return {Socket} for chaining
  * @api public
  */
 
+Socket.prototype.destroy =
 Socket.prototype.close = function () {
   if ('open' == this.readyState) {
     this.readyState = 'closing';
@@ -362,4 +414,19 @@ Socket.prototype.close = function () {
       self.onClose('forced close');
     });
   }
+  return this;
 };
+
+
+/**
+ * Closes the socket and underlying transport once the queue is drained.
+ *
+ * @return {Socket} for chaining
+ * @api public
+ */
+
+Socket.prototype.destroySoon = function () {
+  this.on('drain', this.close.bind(this));
+  this.flush();
+  return this;
+};

package.json

@@ -26,6 +26,7 @@
       , "superagent": "*"
       , "engine.io-client": "0.4.3"
       , "s": "*"
+      , "morestreams": "*"
     }
   , "scripts" : { "test" : "make test" }
   , "repository": {

test/server.js

@@ -5,7 +5,9 @@
  */
 
 var http = require('http')
-  WebSocket = require('ws');
+  WebSocket = require('ws')
+  fs = require('fs')
+  streams = require('morestreams');
 
 /**
  * Tests.
@@ -1121,4 +1123,126 @@ describe('server', function () {
     });
   });
 
+  describe('stream', function () {
+    it('is pipable into', function (done) {
+      var engine = listen({}, function (port) {
+        var fileSizeOnServer;
+        engine.on('connection', function (conn) {
+          // send this test file and compare the received size
+          fs.createReadStream(__filename).pipe(conn);
+          fs.stat(__filename, function (err, stats) {
+            fileSizeOnServer = stats.size;
+          });
+        });
+        var socket = new eioc.Socket('ws://localhost:%d'.s(port));
+        socket.on('open', function () {
+          var dataLength = 0;
+          socket.on('data', function (data) {
+            dataLength += data.length;
+          });
+          socket.on('close', function () {
+            expect(dataLength).to.be(fileSizeOnServer);
+            done();
+          });
+        });
+      });
+    });
+
+    it('is pipable from', function (done) {
+      var buffered = new streams.BufferedStream();
+      // create simple non-buffered stream, bufferring requires outgoing pipe
+      delete buffered.chunks;
+
+      var engine = listen({}, function (port) {
+        engine.on('connection', function (conn) {
+          conn.pipe(buffered);
+        });
+        var socket = new eioc.Socket('ws://localhost:%d'.s(port));
+        socket.on('open', function () {
+          for(var i = 0; i < 10; i++) {
+            socket.send(i);
+          }
+        });
+        setTimeout(function () {
+          socket.close();
+        }, 1000);
+        var received = "";
+        buffered.on('data', function (data) {
+          received += data;
+        });
+        buffered.on('end', function() {
+          expect(received).to.eql("0123456789");
+          done();
+        });
+      });
+    });
+
+    it('should trigger error on `ping timeout`', function (done) {
+      var opts = { allowUpgrades: false, pingInterval: 5, pingTimeout: 5 };
+      var engine = listen(opts, function (port) {
+        var socket = new eioc.Socket('http://localhost:%d'.s(port));
+        socket.sendPacket = function (){};
+        engine.on('connection', function (conn) {
+          conn.on('error', function (err) {
+            done();
+          })
+        });
+      });
+    });
+
+    it('should trigger error on `transport error`', function ($done) {
+      var engine = listen({ allowUpgrades: false }, function (port) {
+        // hack to access the sockets created by node-xmlhttprequest
+        // see: https://github.com/driverdan/node-XMLHttpRequest/issues/44
+        var request = require('http').request;
+        var sockets = [];
+        http.request = function(opts){
+          var req = request.apply(null, arguments);
+          req.on('socket', function(socket){
+            sockets.push(socket);
+          });
+          return req;
+        };
+
+        function done(){
+          http.request = request;
+          $done();
+        }
+
+        var socket = new eioc.Socket('ws://localhost:%d'.s(port))
+          , serverSocket;
+
+        engine.on('connection', function(s){
+          serverSocket = s;
+        });
+
+        socket.transport.on('poll', function(){
+          // we set a timer to wait for the request to actually reach
+          setTimeout(function(){
+            // kill the underlying connection
+            sockets[1].end();
+            serverSocket.on('error', function(err){
+              expect(err.message).to.be('poll connection closed prematurely');
+              done();
+            });
+          }, 50);
+        });
+      });
+    });
+
+    it('should trigger error on `parse error`', function (done) {
+      var engine = listen({}, function (port) {
+        engine.on('connection', function (conn) {
+          conn.on('error', function (err) {
+            done();
+          });
+        });
+        var socket = new eioc.Socket('ws://localhost:%d'.s(port));
+        socket.on('open', function () {
+          // send bad badly parsed packet
+          socket.sendPacket('error', 'ignored');
+        });
+      });
+    });
+  });
 });