Keep track of when a socket has been destroyed

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 93 matching lines @@
   // eventhandler. When receiving a message from the eventhandler the
   // EVENT flags indicate the events that actually happened. The
   // COMMAND flags are used to send commands from dart to the
   // eventhandler. COMMAND flags are never received from the
   // eventhandler. Additional flags are used to communicate other
   // information.
   static const int READ_EVENT = 0;
   static const int WRITE_EVENT = 1;
   static const int ERROR_EVENT = 2;
   static const int CLOSED_EVENT = 3;
+  static const int DESTROYED_EVENT = 4;
   static const int FIRST_EVENT = READ_EVENT;
-  static const int LAST_EVENT = CLOSED_EVENT;
+  static const int LAST_EVENT = DESTROYED_EVENT;
   static const int EVENT_COUNT = LAST_EVENT - FIRST_EVENT + 1;
 
   static const int CLOSE_COMMAND = 8;
   static const int SHUTDOWN_READ_COMMAND = 9;
   static const int SHUTDOWN_WRITE_COMMAND = 10;
   static const int FIRST_COMMAND = CLOSE_COMMAND;
   static const int LAST_COMMAND = SHUTDOWN_WRITE_COMMAND;
 
   // Type flag send to the eventhandler providing additional
   // information on the type of the file descriptor.
   static const int LISTENING_SOCKET = 16;
   static const int PIPE_SOCKET = 17;
   static const int TYPE_NORMAL_SOCKET = 0;
   static const int TYPE_LISTENING_SOCKET = 1 << LISTENING_SOCKET;
   static const int TYPE_PIPE = 1 << PIPE_SOCKET;
 
   // Native port messages.
   static const HOST_NAME_LOOKUP = 0;
 
   // Socket close state
   bool isClosed = false;
+  bool isClosing = false;
   bool isClosedRead = false;
   bool isClosedWrite = false;
   Completer closeCompleter = new Completer();
 
   // Handlers and receive port for socket events from the event handler.
   int eventMask = 0;
   List eventHandlers;
   ReceivePort eventPort;
 
   // Indicates if native interrupts can be activated.
@@ skipping 107 matching lines @@
     eventHandlers = new List(EVENT_COUNT + 1);
     _EventHandler._start();
   }
 
   _NativeSocket.pipe() : typeFlags = TYPE_PIPE {
     eventHandlers = new List(EVENT_COUNT + 1);
     _EventHandler._start();
   }
 
   int available() {
-    if (isClosed) return 0;
+    if (isClosing || isClosed) return 0;
     var result = nativeAvailable();
     if (result is OSError) {
       reportError(result, "Available failed");
       return 0;
     } else {
       return result;
     }
   }
 
   List<int> read(int len) {
     if (len != null && len <= 0) {
       throw new ArgumentError("Illegal length $len");
     }
-    if (isClosed) return null;
+    if (isClosing || isClosed) return null;
     var result = nativeRead(len == null ? -1 : len);
     if (result is OSError) {
       reportError(result, "Read failed");
       return null;
     }
     return result;
   }
 
   int write(List<int> buffer, int offset, int bytes) {
     if (buffer is! List) throw new ArgumentError();
     if (offset == null) offset = 0;
     if (bytes == null) bytes = buffer.length;
     if (offset < 0) throw new RangeError.value(offset);
     if (bytes < 0) throw new RangeError.value(bytes);
     if ((offset + bytes) > buffer.length) {
       throw new RangeError.value(offset + bytes);
     }
     if (offset is! int || bytes is! int) {
       throw new ArgumentError("Invalid arguments to write on Socket");
     }
-    if (isClosed) return 0;
+    if (isClosing || isClosed) return 0;
     if (bytes == 0) return 0;
     _BufferAndStart bufferAndStart =
         _ensureFastAndSerializableBuffer(buffer, offset, offset + bytes);
     var result =
         nativeWrite(bufferAndStart.buffer, bufferAndStart.start, bytes);
     if (result is OSError) {
       reportError(result, "Write failed");
       result = 0;
     }
     return result;
@@ skipping 20 matching lines @@
     return nativeGetRemotePeer()[0];
   }
 
   // Multiplexes socket events to the socket handlers.
   void multiplex(int events) {
     canActivateEvents = false;
     for (int i = FIRST_EVENT; i <= LAST_EVENT; i++) {
       if (((events & (1 << i)) != 0)) {
         if (i == CLOSED_EVENT &&
             typeFlags != TYPE_LISTENING_SOCKET &&
+            !isClosing &&
             !isClosed) {
           isClosedRead = true;
         }
 
         var handler = eventHandlers[i];
+        if (i == DESTROYED_EVENT) {
+          assert(!isClosed);
+          isClosed = true;
+          closeCompleter.complete(this);
+          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
           // no need to notify the eventhandler as handlers are
           // disabled while the event is handled.
           eventMask &= ~(1 << i);
         }
@@ skipping 10 matching lines @@
         } else if (!isClosed) {
           handler();
         }
       }
     }
     if (isClosedRead && isClosedWrite) close();
     canActivateEvents = true;
     activateHandlers();
   }
 
-  void setHandlers({read: null, write: null, error: null, closed: null}) {
+  void setHandlers({read, write, error, closed, destroyed}) {
     eventHandlers[READ_EVENT] = read;
     eventHandlers[WRITE_EVENT] = write;
     eventHandlers[ERROR_EVENT] = error;
     eventHandlers[CLOSED_EVENT] = closed;
+    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 get closeFuture => closeCompleter.future;
+  Future<_NativeSocket> get closeFuture => closeCompleter.future;
 
   void activateHandlers() {
     if (canActivateEvents && !isClosed) {
-      // If we don't listen for either read or write, disconnect as we won't
-      // get close and error events anyway.
-      if ((eventMask & ((1 << READ_EVENT) | (1 << WRITE_EVENT))) == 0) {
+      if (!isClosing &&
+          (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 (isClosing) {
+          data = 1 << DESTROYED_EVENT;
+        } else {
+          if (isClosedRead) data &= ~(1 << READ_EVENT);
+          if (isClosedWrite) data &= ~(1 << WRITE_EVENT);
+        }
         data |= typeFlags;
-        if (isClosedRead) data &= ~(1 << READ_EVENT);
-        if (isClosedWrite) data &= ~(1 << WRITE_EVENT);
         sendToEventHandler(data);
       }
     }
   }
 
-  void close() {
-    if (!isClosed) {
+  Future<_NativeSocket> close() {
+    if (!isClosing && !isClosed) {
       sendToEventHandler(1 << CLOSE_COMMAND);
-      isClosed = true;
-      closeCompleter.complete(this);
+      isClosing = true;
     }
-    // Outside the if support closing sockets created but never
-    // assigned any actual socket.
-    disconnectFromEventHandler();
+    return closeFuture;
   }
 
   void shutdown(SocketDirection direction) {
-    if (!isClosed) {
+    if (!isClosing && !isClosed) {
       switch (direction) {
         case SocketDirection.RECEIVE:
           shutdownRead();
           break;
         case SocketDirection.SEND:
           shutdownWrite();
           break;
         case SocketDirection.BOTH:
           close();
           break;
         default:
           throw new ArgumentError(direction);
       }
     }
   }
 
   void shutdownWrite() {
-    if (!isClosed) {
+    if (!isClosing && !isClosed) {
       if (isClosedRead) {
         close();
       } else {
         sendToEventHandler(1 << SHUTDOWN_WRITE_COMMAND);
       }
       isClosedWrite = true;
     }
   }
 
   void shutdownRead() {
-    if (!isClosed) {
+    if (!isClosing && !isClosed) {
       if (isClosedWrite) {
         close();
       } else {
         sendToEventHandler(1 << SHUTDOWN_READ_COMMAND);
       }
       isClosedRead = true;
     }
   }
 
   void sendToEventHandler(int data) {
@@ skipping 123 matching lines @@
                                         bool cancelOnError}) {
     return _controller.stream.listen(
         onData,
         onError: onError,
         onDone: onDone,
         cancelOnError: cancelOnError);
   }
 
   int get port => _socket.port;
 
-  void close() => _socket.close();
+  Future close() => _socket.close().then((_) => this);
 
   void _pause() {
     _socket.setListening(read: false, write: false);
   }
 
   void _resume() {
     _socket.setListening(read: true, write: false);
   }
 
   void _onSubscriptionStateChange() {
@@ skipping 35 matching lines @@
     _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),
       error: (e) {
         _controller.addError(e);
         close();
       }
     );
   }
 
   factory _RawSocket._writePipe(int fd) {
     var native = new _NativeSocket.pipe();
     native.isClosedRead = true;
@@ skipping 19 matching lines @@
         cancelOnError: cancelOnError);
   }
 
   int available() => _socket.available();
 
   List<int> read([int len]) => _socket.read(len);
 
   int write(List<int> buffer, [int offset, int count]) =>
       _socket.write(buffer, offset, count);
 
-  void close() => _socket.close();
+  Future close() => _socket.close().then((_) => this);
 
   void shutdown(SocketDirection direction) => _socket.shutdown(direction);
 
   int get port => _socket.port;
 
   int get remotePort => _socket.remotePort;
 
   InternetAddress get address => _socket.address;
 
   String get remoteHost => _socket.remoteHost;
@@ skipping 72 matching lines @@
                                      bool cancelOnError}) {
     return _socket.map((rawSocket) => new _Socket(rawSocket)).listen(
         onData,
         onError: onError,
         onDone: onDone,
         cancelOnError: cancelOnError);
   }
 
   int get port => _socket.port;
 
-  void close() => _socket.close();
+  Future close() => _socket.close().then((_) => this);
 }
 
 
 patch class Socket {
   /* patch */ static Future<Socket> connect(host, int port) {
     return RawSocket.connect(host, port).then(
         (socket) => new _Socket(socket));
   }
 }
 
@@ skipping 159 matching lines @@
   void add(List<int> bytes) => _sink.add(bytes);
 
   Future<Socket> addStream(Stream<List<int>> stream) {
     return _sink.addStream(stream);
   }
 
   Future<Socket> close() => _sink.close();
 
   Future<Socket> get done => _sink.done;
 
   void destroy() {

[Anders Johnsen, 2013/05/08 13:11:47]

return Future<Socket>.

     // Destroy can always be called to get rid of a socket.
     if (_raw == null) return;
     _consumer.stop();
     _closeRawSocket();
     _controllerClosed = true;
     _controller.close();
   }
 
   bool setOption(SocketOption option, bool enabled) {
     if (_raw == null) return false;
@@ skipping 124 matching lines @@
     _raw.onBadCertificate = callback;
   }
 
   X509Certificate get peerCertificate {
     if (_raw == null) {
      throw new StateError("peerCertificate called on destroyed SecureSocket");
     }
     return _raw.peerCertificate;
   }
 }