Catch errors from stream events in net.js

Pipe into 'error' event.

lib/net.js

@@ -267,8 +267,13 @@ function _doFlush () {
   // Stream becomes writeable on connect() but don't flush if there's
   // nothing actually to write
   if (socket.flush()) {
+    try {
       if (socket._events && socket._events['drain']) socket.emit("drain");
       if (socket.ondrain) socket.ondrain(); // Optimization
+    } catch (e) {
+      socket.destroy(e);
+      return;
+    }
   }
 }
 
@@ -320,7 +325,7 @@ function initStream (self) {
 
     //debug('bytesRead ' + bytesRead + '\n');
 
-    if (self.secure && bytesRead == 0 && secureBytesRead > 0){
+    if (self.secure && bytesRead == 0 && secureBytesRead > 0) {
       // Deal with SSL handshake
       if (self.server) {
         self._checkForSecureHandshake();
@@ -338,8 +343,13 @@ function initStream (self) {
       if (!self.writable) self.destroy();
       // Note: 'close' not emitted until nextTick.
 
+      try {
         if (self._events && self._events['end']) self.emit('end');
         if (self.onend) self.onend();
+      } catch (e) {
+        self.destroy(e);
+        return;
+      }
     } else if (bytesRead > 0) {
 
       timeout.active(self);
@@ -348,6 +358,7 @@ function initStream (self) {
       var end = pool.used + bytesRead;
       pool.used += bytesRead;
 
+      try {
         if (!self._encoding) {
           if (self._events && self._events['data']) {
             // emit a slice
@@ -360,6 +371,10 @@ function initStream (self) {
           var string = pool.toString(self._encoding, start, end);
           self.emit('data', string);
         }
+      } catch (e) {
+        self.destroy(e);
+        return;
+      }
     }
   };
   self.readable = false;
@@ -586,9 +601,14 @@ Stream.prototype._writeOut = function (data, encoding) {
       } else {
         var secureBytesWritten = write(this.fd, securePool, 0, secureLen);
       }
-      if(!this.secureEstablished && this.secureStream.isInitFinished()) {
+      if (!this.secureEstablished && this.secureStream.isInitFinished()) {
         this.secureEstablished = true;
+        try {
           if (this._events && this._events['secure']) this.emit('secure');
+        } catch (e) {
+          this.destroy(e);
+          return;
+        }
       }
     } else {
       bytesWritten = write(this.fd, buffer, off, len);
@@ -696,7 +716,12 @@ function doConnect (socket, port, host) {
       socket.resume();
       socket.readable = socket.writable = true;
       socket._writeWatcher.callback = _doFlush;
+      try {
         socket.emit('connect');
+      } catch (e) {
+        socket.destroy(e);
+        return;
+      }
     } else if (errno != EINPROGRESS) {
       socket.destroy(errnoException(errno, 'connect'));
     }
@@ -902,9 +927,15 @@ function Server (listener) {
       s.resume();
 
       self.emit('connection', s);
+
       // The 'connect' event  probably should be removed for server-side
       // sockets. It's redundant.
+      try {
         s.emit('connect');
+      } catch (e) {
+        s.destroy(e);
+        return;
+      }
     }
   };
 }