Avoid over-use of futures in socket.

runtime/bin/socket_patch.dart

 // Copyright (c) 2013, 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.
 
 patch class RawServerSocket {
   /* patch */ static Future<RawServerSocket> bind(address,
                                                   int port,
                                                   {int backlog: 0,
                                                    bool v6Only: false}) {
     return _RawServerSocket.bind(address, port, backlog, v6Only);
@@ skipping 235 matching lines @@
 
   // Protocol flags.
   static const int PROTOCOL_IPV4 = 1 << 0;
   static const int PROTOCOL_IPV6 = 1 << 1;
 
   // Socket close state
   bool isClosed = false;
   bool isClosing = false;
   bool isClosedRead = false;
   bool isClosedWrite = false;
-  Completer closeCompleter = new Completer();
+  Completer closeCompleter = new Completer.sync();
 
   // Handlers and receive port for socket events from the event handler.
   int eventMask = 0;
   List eventHandlers;
   RawReceivePort eventPort;
 
   // Indicates if native interrupts can be activated.
   bool canActivateEvents = true;
 
   // The type flags for this socket.
@@ skipping 297 matching lines @@
             typeFlags != TYPE_LISTENING_SOCKET &&
             !isClosing &&
             !isClosed) {
           isClosedRead = true;
         }
 
         var handler = eventHandlers[i];
         if (i == DESTROYED_EVENT) {
           assert(!isClosed);
           isClosed = true;
-          closeCompleter.complete(this);
+          closeCompleter.complete();
           disconnectFromEventHandler();
           if (handler != null) handler();
           continue;
         }
         assert(handler != null);
         if (i == WRITE_EVENT) {
           // If the event was disabled before we had a chance to fire the event,
           // discard it. If we register again, we'll get a new one.
           if ((eventMask & (1 << i)) == 0) continue;
           // Unregister the out handler before executing it. There is
@@ skipping 33 matching lines @@
     eventHandlers[DESTROYED_EVENT] = destroyed;
   }
 
   void setListening({read: true, write: true}) {
     eventMask = (1 << CLOSED_EVENT) | (1 << ERROR_EVENT);
     if (read) eventMask |= (1 << READ_EVENT);
     if (write) eventMask |= (1 << WRITE_EVENT);
     activateHandlers();
   }
 
-  Future<_NativeSocket> get closeFuture => closeCompleter.future;
 
   void activateHandlers() {
     if (canActivateEvents && !isClosing && !isClosed) {
       if ((eventMask & ((1 << READ_EVENT) | (1 << WRITE_EVENT))) == 0) {
         // If we don't listen for either read or write, disconnect as we won't
         // get close and error events anyway.
         if (eventPort != null) disconnectFromEventHandler();
       } else {
         int data = eventMask;
         if (isClosedRead) data &= ~(1 << READ_EVENT);
         if (isClosedWrite) data &= ~(1 << WRITE_EVENT);
         data |= typeFlags;
         sendToEventHandler(data);
       }
     }
   }
 
-  Future<_NativeSocket> close() {
+  Future close() {
     if (!isClosing && !isClosed) {
       sendToEventHandler(1 << CLOSE_COMMAND);
       isClosing = true;
     }
-    return closeFuture;
+    return closeCompleter.future;
   }
 
   void shutdown(SocketDirection direction) {
     if (!isClosing && !isClosed) {
       switch (direction) {
         case SocketDirection.RECEIVE:
           shutdownRead();
           break;
         case SocketDirection.SEND:
           shutdownWrite();
@@ skipping 198 matching lines @@
         .then((socket) => new _RawServerSocket(socket));
   }
 
   _RawServerSocket(this._socket) {
     var zone = Zone.current;
     _controller = new StreamController(sync: true,
         onListen: _onSubscriptionStateChange,
         onCancel: _onSubscriptionStateChange,
         onPause: _onPauseStateChange,
         onResume: _onPauseStateChange);
-    _socket.closeFuture.then((_) => _controller.close());
     _socket.setHandlers(
       read: zone.bindCallback(() {
         var socket = _socket.accept();
         if (socket != null) _controller.add(new _RawSocket(socket));
       }),
       error: zone.bindUnaryCallback((e) {
         _controller.addError(e);
         _controller.close();
-      })
+      }),
+      destroyed: _controller.close
     );
   }
 
   StreamSubscription<RawSocket> listen(void onData(RawSocket event),
                                        {Function onError,
                                         void onDone(),
                                         bool cancelOnError}) {
     return _controller.stream.listen(
         onData,
         onError: onError,
@@ skipping 12 matching lines @@
   }
 
   void _resume() {
     _socket.setListening(read: true, write: false);
   }
 
   void _onSubscriptionStateChange() {
     if (_controller.hasListener) {
       _resume();
     } else {
-      close();
+      _socket.close();
     }
   }
 
   void _onPauseStateChange() {
     if (_controller.isPaused) {
       _pause();
     } else {
       _resume();
     }
   }
@@ skipping 15 matching lines @@
         .then((socket) => new _RawSocket(socket));
   }
 
   _RawSocket(this._socket) {
     var zone = Zone.current;
     _controller = new StreamController(sync: true,
         onListen: _onSubscriptionStateChange,
         onCancel: _onSubscriptionStateChange,
         onPause: _onPauseStateChange,
         onResume: _onPauseStateChange);
-    _socket.closeFuture.then((_) => _controller.close());
     _socket.setHandlers(
       read: () => _controller.add(RawSocketEvent.READ),
       write: () {
         // The write event handler is automatically disabled by the
         // event handler when it fires.
         _writeEventsEnabled = false;
         _controller.add(RawSocketEvent.WRITE);
       },
       closed: () => _controller.add(RawSocketEvent.READ_CLOSED),
-      destroyed: () => _controller.add(RawSocketEvent.CLOSED),
+      destroyed: () {
+        _controller.add(RawSocketEvent.CLOSED);
+        _controller.close();
+      },
       error: zone.bindUnaryCallback((e) {
         _controller.addError(e);
-        close();
+        _socket.close();
       })
     );
   }
 
   factory _RawSocket._writePipe(int fd) {
     var native = new _NativeSocket.pipe();
     native.isClosedRead = true;
     if (fd != null) _getStdioHandle(native, fd);
     return new _RawSocket(native);
   }
@@ skipping 85 matching lines @@
       _pause();
     } else {
       _resume();
     }
   }
 
   void _onSubscriptionStateChange() {
     if (_controller.hasListener) {
       _resume();
     } else {
-      close();
+      _socket.close();
     }
   }
 }
 
 
 patch class ServerSocket {
   /* patch */ static Future<ServerSocket> bind(address,
                                                int port,
                                                {int backlog: 0,
                                                 bool v6Only: false}) {
@@ skipping 342 matching lines @@
   bool _readEventsEnabled = true;
   bool _writeEventsEnabled = true;
 
   _RawDatagramSocket(this._socket) {
     var zone = Zone.current;
     _controller = new StreamController(sync: true,
         onListen: _onSubscriptionStateChange,
         onCancel: _onSubscriptionStateChange,
         onPause: _onPauseStateChange,
         onResume: _onPauseStateChange);
-    _socket.closeFuture.then((_) => _controller.close());
     _socket.setHandlers(
       read: () => _controller.add(RawSocketEvent.READ),
       write: () {
         // The write event handler is automatically disabled by the
         // event handler when it fires.
         _writeEventsEnabled = false;
         _controller.add(RawSocketEvent.WRITE);
       },
       closed: () => _controller.add(RawSocketEvent.READ_CLOSED),
-      destroyed: () => _controller.add(RawSocketEvent.CLOSED),
+      destroyed: () {
+        _controller.add(RawSocketEvent.CLOSED);
+        _controller.close();
+      },
       error: zone.bindUnaryCallback((e) {
         _controller.addError(e);
-        close();
+        _socket.close();
       })
     );
   }
 
   static Future<RawDatagramSocket> bind(
       host, int port, bool reuseAddress) {
     if (port < 0 || port > 0xffff)
       throw new ArgumentError("Invalid port $port");
     return _NativeSocket.bindDatagram(host, port, reuseAddress)
         .then((socket) => new _RawDatagramSocket(socket));
@@ skipping 80 matching lines @@
       _pause();
     } else {
       _resume();
     }
   }
 
   void _onSubscriptionStateChange() {
     if (_controller.hasListener) {
       _resume();
     } else {
-      close();
+      _socket.close();
     }
   }
 }
 
 Datagram _makeDatagram(List<int> data,
                        String address,
                        List<int> in_addr,
                        int port) {
   return new Datagram(
       data,
       new _InternetAddress(address, null, in_addr),
       port);
 }