initial port to streams2

brianc · brianc · commit 00c3f26c295d · 2013-04-05T10:00:47.000-05:00
diff --git a/lib/connection.js b/lib/connection.js
@@ -3,6 +3,8 @@ var crypto = require('crypto');
 var EventEmitter = require('events').EventEmitter;
 var util = require('util');
 
+var Readable = require('stream').Readable;
+
 var utils = require(__dirname + '/utils');
 var Writer = require('buffer-writer');
 
@@ -37,15 +39,6 @@ Connection.prototype.connect = function(port, host) {
     self.emit('connect');
   });
 
-  this.stream.on('error', function(error) {
-    //don't raise ECONNRESET errors - they can & should be ignored 
-    //during disconnect
-    if(self._ending && error.code == 'ECONNRESET') {
-      return;
-    }
-    self.emit('error', error);
-  });
-
   this.stream.on('end', function() {
     self.emit('end');
   });
@@ -89,15 +82,27 @@ Connection.prototype.connect = function(port, host) {
 
 Connection.prototype.attachListeners = function(stream) {
   var self = this;
-  stream.on('data', function(buffer) {
-    self.setBuffer(buffer);
+  var reader = new Readable();
+  reader.wrap(stream);
+  reader.on('readable', function() {
+    self.setBuffer(reader.read());
     var msg = self.parseMessage();
     while(msg) {
       self.emit('message', msg);
       self.emit(msg.name, msg);
       msg = self.parseMessage();
     }
   });
+  reader.on('error', function(error) {
+    //don't raise ECONNRESET errors - they can & should be ignored 
+    //during disconnect
+    if(self._ending && error.code == 'ECONNRESET') {
+      return;
+    }
+    self.emit('error', error);
+  });
+
+  reader.read(0);
 };
 
 Connection.prototype.requestSsl = function(config) {
