Make server-sockets level triggered.

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 394 matching lines @@
   final int typeFlags;
 
   // Holds the port of the socket, 0 if not known.
   int localPort = 0;
 
   // Holds the address used to connect or bind the socket.
   InternetAddress address;
 
   int available = 0;
 
-  int returnTokens = 0;
+  int tokens = 0;
 
   bool sendReadEvents = false;
   bool readEventIssued = false;
 
   bool sendWriteEvents = false;
   bool writeEventIssued = false;
   bool writeAvailable = false;
 
   // Statistics.
   int totalRead = 0;
@@ skipping 263 matching lines @@
     if (result is OSError) {
       scheduleMicrotask(() => reportError(result, "Send failed"));
       result = 0;
     }
     return result;
   }
 
   _NativeSocket accept() {
     // Don't issue accept if we're closing.
     if (isClosing || isClosed) return null;
+    assert(available > 0);
+    available--;
+    tokens++;
+    returnTokens();
     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 72 matching lines @@
           issueReadEvent();
           continue;
         }
 
         if (i == WRITE_EVENT) {
           writeAvailable = true;
           issueWriteEvent(delayed: false);
           continue;
         }
 
-        if (i == READ_EVENT && !isListening) {
-          var avail = nativeAvailable();
-          if (avail is int) {
-            available = avail;
+        if (i == READ_EVENT) {
+          if (isListening) {
+            available++;
           } else {
-            // Available failed. Mark socket as having data, to ensure read
-            // events, and thus reporting of this error.
-            available = 1;
+            var avail = nativeAvailable();
+            if (avail is int) {
+              available = avail;
+            } else {
+              // Available failed. Mark socket as having data, to ensure read
+              // events, and thus reporting of this error.
+              available = 1;
+            }
+            issueReadEvent();
+            continue;
           }
-          issueReadEvent();
-          continue;
         }
 
         var handler = eventHandlers[i];
         if (i == DESTROYED_EVENT) {
           assert(!isClosed);
           isClosed = true;
           closeCompleter.complete();
           disconnectFromEventHandler();
           if (handler != null) handler();
           continue;
         }
 
         if (i == ERROR_EVENT) {
           if (!isClosing) {
             reportError(nativeGetError(), "");
           }
         } 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 (eventPort != null && !isClosing && !isClosed && !isListening) {
-      returnTokens++;
-      if (returnTokens == 8) {
+    if (!isListening) {
+      tokens++;
+      returnTokens();
+    }
+  }
+
+  void returnTokens() {
+    if (eventPort != null && !isClosing && !isClosed) {
+      if (tokens == 8) {
         // Return in batches of 8.
-        assert(returnTokens < (1 << FIRST_COMMAND));
-        sendToEventHandler((1 << RETURN_TOKEN_COMMAND) | returnTokens);
-        returnTokens = 0;
+        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;
     eventHandlers[ERROR_EVENT] = error;
     eventHandlers[CLOSED_EVENT] = closed;
     eventHandlers[DESTROYED_EVENT] = destroyed;
@@ skipping 236 matching lines @@
       throw new StateError("Stream was already listened to");
     }
     var zone = Zone.current;
     _controller = new StreamController(sync: true,
         onListen: _onSubscriptionStateChange,
         onCancel: _onSubscriptionStateChange,
         onPause: _onPauseStateChange,
         onResume: _onPauseStateChange);
     _socket.setHandlers(
       read: zone.bindCallback(() {
-        do {
+        while (_socket.available > 0) {
           var socket = _socket.accept();
           if (socket == null) return;
           _controller.add(new _RawSocket(socket));
-        } while (!_controller.isPaused);
+          if (_controller.isPaused) return;
+        }
       }),
       error: zone.bindUnaryCallback((e) {
         _controller.addError(e);
         _controller.close();
       }),
       destroyed: _controller.close);
     return _controller.stream.listen(
         onData,
         onError: onError,
         onDone: onDone,
@@ skipping 662 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();