Add error handling to socket streams

runtime/bin/socket_impl.dart

 // Copyright (c) 2012, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 
 class _SocketBase {
   // Bit flags used when communicating between the eventhandler and
   // dart code. The EVENT flags are used to indicate events of
   // interest when sending a message from dart code to the
   // eventhandler. When receiving a message from the eventhandler the
@@ skipping 36 matching lines @@
   void _multiplex(int event_mask) {
     _canActivateHandlers = false;
     for (int i = _FIRST_EVENT; i <= _LAST_EVENT; i++) {
       if (((event_mask & (1 << i)) != 0)) {
         if ((i == _CLOSE_EVENT) && this is _Socket && _id >= 0) {
           _closedRead = true;
           if (_closedWrite) _close();
         }
 
         var eventHandler = _handlerMap[i];
-        if (eventHandler != null) {
+        if (eventHandler != null || i == _ERROR_EVENT) {
           // Unregister the out handler before executing it.
           if (i == _OUT_EVENT) _setHandler(i, null);
 
           // Don't call the in handler if there is no data available
           // after all.
           if ((i == _IN_EVENT) && (this is _Socket) && (available() == 0)) {
             continue;
           }
           if (i == _ERROR_EVENT) {
-            eventHandler(new SocketIOException("", _getError()));
+            _reportError(_getError(), "");
             close();
           } else {
             eventHandler();
           }
         }
       }
     }
     _canActivateHandlers = true;
     _activateHandlers();
   }
@@ skipping 102 matching lines @@
   void _sendToEventHandler(int data) {
     if (_handler === null) {
       _handler = new ReceivePort();
       _handler.receive((var message, ignored) { _multiplex(message); });
     }
     assert(_id >= 0);
     _EventHandler._sendData(_id, _handler, data);
   }
 
   bool _reportError(error, String message) {
+    void doReportError(Exception e) {
+      // Invoke the error callback if any.
+      if (_handlerMap[_ERROR_EVENT] != null) {
+        _handlerMap[_ERROR_EVENT](e);
+      }
+      // Propagate the error to any additional listeners.
+      _propagateError(e);
+    }
+
     // For all errors we close the socket, call the error handler and
     // disable further calls of the error handler.
     close();
-    var onError = _handlerMap[_ERROR_EVENT];
-    if (onError != null) {
-      if (error is OSError) {
-        onError(new SocketIOException(message, error));
-      } else if (error is List) {
-        assert(_isErrorResponse(error));
-        switch (error[0]) {
-          case _FileUtils.kIllegalArgumentResponse:
-            onError(new IllegalArgumentException());
-            break;
-          case _FileUtils.kOSErrorResponse:
-            onError(new SocketIOException(
-                message, new OSError(error[2], error[1])));
-            break;
-          default:
-            onError(new Exception("Unknown error"));
-            break;
-        }
-      } else {
-        onError(new SocketIOException(message));
+    if (error is OSError) {
+      doReportError(new SocketIOException(message, error));
+    } else if (error is List) {
+      assert(_isErrorResponse(error));
+      switch (error[0]) {
+        case _FileUtils.kIllegalArgumentResponse:
+          doReportError(new IllegalArgumentException());
+          break;
+        case _FileUtils.kOSErrorResponse:
+          doReportError(new SocketIOException(
+              message, new OSError(error[2], error[1])));
+          break;
+        default:
+          doReportError(new Exception("Unknown error"));
+          break;
       }
+    } else {
+      doReportError(new SocketIOException(message));
     }
   }
 
   int hashCode() => _hashCode;
 
+  void _propagateError(Exception e) => null;
+
   abstract bool _isListenSocket();
   abstract bool _isPipe();
 
   // Socket id is set from native. -1 indicates that the socket was closed.
   int _id;
 
   // Dedicated ReceivePort for socket events.
   ReceivePort _handler;
 
   // Poll event to handler map.
@@ skipping 228 matching lines @@
     if (_inputStream != null) throw new StreamException(
            "Cannot set close handler when input stream is used");
     _onClosed = callback;
   }
 
   bool _isListenSocket() => false;
 
   bool _isPipe() => _pipe;
 
   InputStream get inputStream() {
-    if (_inputStream === null) {
+    if (_inputStream == null) {
       if (_handlerMap[_IN_EVENT] !== null ||
           _handlerMap[_CLOSE_EVENT] !== null) {
         throw new StreamException(
             "Cannot get input stream when socket handlers are used");
       }
       _inputStream = new SocketInputStream(this);
     }
     return _inputStream;
   }
 
   OutputStream get outputStream() {
-    if (_outputStream === null) {
+    if (_outputStream == null) {
       if (_handlerMap[_OUT_EVENT] !== null) {
         throw new StreamException(
             "Cannot get input stream when socket handlers are used");
       }
       _outputStream = new SocketOutputStream(this);
     }
     return _outputStream;
   }
 
   void set _onWrite(void callback()) {
     _setHandler(_OUT_EVENT, callback);
   }
 
   void set _onData(void callback()) {
     _setHandler(_IN_EVENT, callback);
   }
 
   void set _onClosed(void callback()) {
     _setHandler(_CLOSE_EVENT, callback);
   }
 
+  void _propagateError(Exception e) {
+    if (_inputStream != null) {
+      _inputStream._onError(e);
+    }
+    if (_outputStream != null) {
+      _outputStream._onError(e);
+    }
+  }
+
   void _updateOutHandler() {
     void firstWriteHandler() {
       assert(!_seenFirstOutEvent);
       _seenFirstOutEvent = true;
 
       // From now on the write handler is only the client write
       // handler (connect handler cannot be called again). Change this
       // before calling any handlers as handlers can change the
       // handlers.
       if (_clientWriteHandler === null) _onWrite = _clientWriteHandler;
@@ skipping 26 matching lines @@
   }
 
   bool _seenFirstOutEvent = false;
   bool _pipe = false;
   Function _clientConnectHandler;
   Function _clientWriteHandler;
   SocketInputStream _inputStream;
   SocketOutputStream _outputStream;
   static SendPort _socketService;
 }