Add support for cloning server-sockets.

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 372 matching lines @@
 
   // Native port messages.
   static const HOST_NAME_LOOKUP = 0;
   static const LIST_INTERFACES = 1;
   static const REVERSE_LOOKUP = 2;
 
   // Protocol flags.
   static const int PROTOCOL_IPV4 = 1 << 0;
   static const int PROTOCOL_IPV6 = 1 << 1;
 
+  static const int NORMAL_TOKEN_BATCH_SIZE = 8;
+  static const int LISTENING_TOKEN_BATCH_SIZE = 2;
+
   // Socket close state
   bool isClosed = false;
   bool isClosing = false;
   bool isClosedRead = false;
   bool isClosedWrite = false;
   Completer closeCompleter = new Completer.sync();
 
   // Handlers and receive port for socket events from the event handler.
   final List eventHandlers = new List(EVENT_COUNT + 1);
   RawReceivePort eventPort;
@@ skipping 302 matching lines @@
     }
     return result;
   }
 
   _NativeSocket accept() {
     // Don't issue accept if we're closing.
     if (isClosing || isClosed) return null;
     assert(available > 0);
     available--;
     tokens++;
-    returnTokens();
+    returnTokens(LISTENING_TOKEN_BATCH_SIZE);
     var socket = new _NativeSocket.normal();
     if (nativeAccept(socket) != true) return null;
     socket.localPort = localPort;
     socket.address = address;
     totalRead += 1;
     return socket;
   }
 
   int get port {
     if (localPort != 0) return localPort;
@@ skipping 112 matching lines @@
           }
         } else if (!isClosed) {
           // If the connection is closed right after it's accepted, there's a
           // chance the close-handler is not set.
           if (handler != null) handler();
         }
       }
     }
     if (!isListening) {
       tokens++;
-      returnTokens();
+      returnTokens(NORMAL_TOKEN_BATCH_SIZE);
     }
   }
 
-  void returnTokens() {
+  void returnTokens(int tokenBatchSize) {
     if (eventPort != null && !isClosing && !isClosed) {
-      if (tokens == 8) {
-        // Return in batches of 8.
+      // Return in batches.
+      if (tokens == tokenBatchSize) {
         assert(tokens < (1 << FIRST_COMMAND));
         sendToEventHandler((1 << RETURN_TOKEN_COMMAND) | tokens);
         tokens = 0;
       }
     }
   }
 
   void setHandlers({read, write, error, closed, destroyed}) {
     eventHandlers[READ_EVENT] = read;
     eventHandlers[WRITE_EVENT] = write;
@@ skipping 193 matching lines @@
       native "Socket_SendTo";
   nativeCreateConnect(List<int> addr,
                       int port) native "Socket_CreateConnect";
   nativeCreateBindListen(List<int> addr, int port, int backlog, bool v6Only)
       native "ServerSocket_CreateBindListen";
   nativeCreateBindDatagram(List<int> addr, int port, bool reuseAddress)
       native "Socket_CreateBindDatagram";
   nativeAccept(_NativeSocket socket) native "ServerSocket_Accept";
   int nativeGetPort() native "Socket_GetPort";
   List nativeGetRemotePeer() native "Socket_GetRemotePeer";
+  int nativeGetSocketId() native "Socket_GetSocketId";
   OSError nativeGetError() native "Socket_GetError";
   nativeGetOption(int option, int protocol) native "Socket_GetOption";
   bool nativeSetOption(int option, int protocol, value)
       native "Socket_SetOption";
   bool nativeJoinMulticast(
       List<int> addr, List<int> interfaceAddr, int interfaceIndex)
           native "Socket_JoinMulticast";
   bool nativeLeaveMulticast(
       List<int> addr, List<int> interfaceAddr, int interfaceIndex)
       native "Socket_LeaveMulticast";
 }
 
 
 class _RawServerSocket extends Stream<RawSocket>
                        implements RawServerSocket {
   final _NativeSocket _socket;
   StreamController<RawSocket> _controller;
+  ReceivePort _port;

[Søren Gjesse, 2014/05/06 08:47:39]

Call this variable something less generic, e.g. _referencePort.

[Anders Johnsen, 2014/05/06 12:34:32]

Done.

 
   static Future<_RawServerSocket> bind(address,
                                        int port,
                                        int backlog,
                                        bool v6Only) {
     if (port < 0 || port > 0xFFFF)
       throw new ArgumentError("Invalid port $port");
     if (backlog < 0) throw new ArgumentError("Invalid backlog $backlog");
     return _NativeSocket.bind(address, port, backlog, v6Only)
         .then((socket) => new _RawServerSocket(socket));
@@ skipping 20 matching lines @@
           var socket = _socket.accept();
           if (socket == null) return;
           _controller.add(new _RawSocket(socket));
           if (_controller.isPaused) return;
         }
       }),
       error: zone.bindUnaryCallback((e) {
         _controller.addError(e);
         _controller.close();
       }),
-      destroyed: _controller.close);
+      destroyed: () {
+        _controller.close();
+        if (_port != null) {
+          _port.close();
+        }
+      });
     return _controller.stream.listen(
         onData,
         onError: onError,
         onDone: onDone,
         cancelOnError: cancelOnError);
   }
 
   int get port => _socket.port;
 
   InternetAddress get address => _socket.address;
@@ skipping 16 matching lines @@
     }
   }
 
   void _onPauseStateChange() {
     if (_controller.isPaused) {
       _pause();
     } else {
       _resume();
     }
   }
+
+  RawServerSocketReference get reference {
+    if (_port == null) {
+      _port = new ReceivePort();
+      _port.listen((sendPort) {
+        sendPort.send(
+          [_socket.nativeGetSocketId(),
+           _socket.address,
+           _socket.localPort]);
+      });
+    }
+    return new _RawServerSocketReference(_port.sendPort);
+  }
 }
 
 
+class _RawServerSocketReference implements RawServerSocketReference {
+  final SendPort _sendPort;
+
+  _RawServerSocketReference(this._sendPort);
+
+  Future<RawServerSocket> create() {

[Lasse Reichstein Nielsen, 2014/05/06 09:04:20]

I would still prefer to just return a RawServerSocket.

[Anders Johnsen, 2014/05/06 12:34:32]

The problem is that it may return an invalid object. Let's iterate this one some
more.

+    var port = new ReceivePort();

[Lasse Reichstein Nielsen, 2014/05/06 09:04:20]

Consider using a RawReceivePort. It requires you to do more low-level stuff, but
it probably has lower overhead.
On the other hand, this function isn't likely to be performance critical.

[Anders Johnsen, 2014/05/06 12:34:32]

Done.

+    _sendPort.send(port.sendPort);
+    return port.first.then((args) {
+      port.close();
+      var native = new _NativeSocket.listen();
+      native.nativeSetSocketId(args[0]);
+      native.address = args[1];
+      native.localPort = args[2];
+      return new _RawServerSocket(native);
+    });
+  }

[Lasse Reichstein Nielsen, 2014/05/06 09:04:20]

Consider adding:

  int get hashCode => _sendPort.hashCode;
  bool operator==(Object other) 
      => other is _RawServerSocketReference && 
         sendPort == other._sendPort;

[Anders Johnsen, 2014/05/06 12:34:32]

Done.

+}
+
+
 class _RawSocket extends Stream<RawSocketEvent>
                  implements RawSocket {
   final _NativeSocket _socket;
   StreamController<RawSocketEvent> _controller;
   bool _readEventsEnabled = true;
   bool _writeEventsEnabled = true;
 
   // Flag to handle Ctrl-D closing of stdio on Mac OS.
   bool _isMacOSTerminalInput = false;
 
@@ skipping 137 matching lines @@
 
 patch class ServerSocket {
   /* patch */ static Future<ServerSocket> bind(address,
                                                int port,
                                                {int backlog: 0,
                                                 bool v6Only: false}) {
     return _ServerSocket.bind(address, port, backlog, v6Only);
   }
 }
 
+
+class _ServerSocketReference implements ServerSocketReference {
+  final RawServerSocketReference _rawReference;
+
+  _ServerSocketReference(this._rawReference);
+
+  Future<ServerSocket> create() {
+    return _rawReference.create().then((raw) => new _ServerSocket(raw));
+  }
+}
+
+
 class _ServerSocket extends Stream<Socket>
                     implements ServerSocket {
   final _socket;
 
   static Future<_ServerSocket> bind(address,
                                     int port,
                                     int backlog,
                                     bool v6Only) {
     return _RawServerSocket.bind(address, port, backlog, v6Only)
         .then((socket) => new _ServerSocket(socket));
@@ skipping 10 matching lines @@
         onError: onError,
         onDone: onDone,
         cancelOnError: cancelOnError);
   }
 
   int get port => _socket.port;
 
   InternetAddress get address => _socket.address;
 
   Future close() => _socket.close().then((_) => this);
+
+  ServerSocketReference get reference {
+    return new _ServerSocketReference(_socket.reference);
+  }
 }
 
 
 patch class Socket {
   /* patch */ static Future<Socket> connect(host, int port) {
     return RawSocket.connect(host, port).then(
         (socket) => new _Socket(socket));
   }
 }
 
@@ skipping 430 matching lines @@
                        String address,
                        List<int> in_addr,
                        int port) {
   return new Datagram(
       data,
       new _InternetAddress(address, null, in_addr),
       port);
 }
 
 String _socketsStats() => _SocketsObservatory.toJSON();