Revert "Extract meta data about open sockets into separate class."

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,
                                                    bool shared: false}) {
@@ skipping 262 matching lines @@
   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;
 
   static const Duration _RETRY_DURATION = const Duration(milliseconds: 250);
   static const Duration _RETRY_DURATION_LOOPBACK =
       const Duration(milliseconds: 25);
 
+  // Use default Map so we keep order.
+  static Map<int, _NativeSocket> _sockets = new Map<int, _NativeSocket>();
+
   // Socket close state
   bool isClosed = false;
   bool isClosing = false;
   bool isClosedRead = false;
   bool closedReadEventSent = 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);
@@ skipping 14 matching lines @@
   int tokens = 0;
 
   bool sendReadEvents = false;
   bool readEventIssued = false;
 
   bool sendWriteEvents = false;
   bool writeEventIssued = false;
   bool writeAvailable = false;
 
   static final Stopwatch sw = new Stopwatch()..start();
-
-  static bool connectedResourceHandler = false;
-  _ReadWriteResourceInfo resourceInfo;
+  // Statistics.
+  int totalRead = 0;
+  int totalWritten = 0;
+  int readCount = 0;
+  int writeCount = 0;
+  double lastRead;
+  double lastWrite;
 
   // The owner object is the object that the Socket is being used by, e.g.
   // a HttpServer, a WebSocket connection, a process pipe, etc.
   Object owner;
 
   static double get timestamp => sw.elapsedMicroseconds / 1000000.0;
 
   static Future<List<InternetAddress>> lookup(
       String host, {InternetAddressType type: InternetAddressType.ANY}) {
     return _IOService._dispatch(_SOCKET_LOOKUP, [host, type._value])
@@ skipping 97 matching lines @@
               connectNext();
             } else {
               // Query the local port, for error messages.
               socket.port;
               // Set up timer for when we should retry the next address
               // (if any).
               var duration = address.isLoopback ?
                   _RETRY_DURATION_LOOPBACK :
                   _RETRY_DURATION;
               var timer = new Timer(duration, connectNext);
-              setupResourceInfo(socket);
-
               connecting[socket] = timer;
               // Setup handlers for receiving the first write event which
               // indicate that the socket is fully connected.
               socket.setHandlers(
                   write: () {
                     timer.cancel();
                     socket.setListening(read: false, write: false);
                     completer.complete(socket);
                     connecting.remove(socket);
                     connecting.forEach((s, t) {
@@ skipping 31 matching lines @@
               .then((list) {
                 if (list.length == 0) {
                   throw createError(response, "Failed host lookup: '$host'");
                 }
                 return list[0];
               });
         })
         .then((address) {
           var socket = new _NativeSocket.listen();
           socket.localAddress = address;
+
           var result = socket.nativeCreateBindListen(address._in_addr,
                                                      port,
                                                      backlog,
                                                      v6Only,
                                                      shared);
           if (result is OSError) {
             throw new SocketException("Failed to create server socket",
                                       osError: result,
                                       address: address,
                                       port: port);
           }
           if (port != 0) socket.localPort = port;
-          setupResourceInfo(socket);
           socket.connectToEventHandler();
           return socket;
         });
   }
 
-  static void setupResourceInfo(_NativeSocket socket) {
-    socket.resourceInfo = new _SocketResourceInfo(socket);
-  }
-
   static Future<_NativeSocket> bindDatagram(
       host, int port, bool reuseAddress) {
     return new Future.value(host)
         .then((host) {
           if (host is _InternetAddress) return host;
           return lookup(host)
               .then((list) {
                 if (list.length == 0) {
                   throw createError(response, "Failed host lookup: '$host'");
                 }
                 return list[0];
               });
         })
         .then((address) {
           var socket = new _NativeSocket.datagram(address);
           var result = socket.nativeCreateBindDatagram(
               address._in_addr, port, reuseAddress);
           if (result is OSError) {
             throw new SocketException("Failed to create datagram socket",
                                       osError: result,
                                       address: address,
                                       port: port);
           }
           if (port != 0) socket.localPort = port;
-          setupResourceInfo(socket);
           return socket;
         });
   }
 
   _NativeSocket.datagram(this.localAddress)
     : typeFlags = TYPE_NORMAL_SOCKET | TYPE_UDP_SOCKET;
 
   _NativeSocket.normal() : typeFlags = TYPE_NORMAL_SOCKET | TYPE_TCP_SOCKET;
 
   _NativeSocket.listen() : typeFlags = TYPE_LISTENING_SOCKET | TYPE_TCP_SOCKET {
@@ skipping 21 matching lines @@
     if (isClosing || isClosed) return null;
     len = min(available, len == null ? available : len);
     if (len == 0) return null;
     var result = nativeRead(len);
     if (result is OSError) {
       reportError(result, "Read failed");
       return null;
     }
     if (result != null) {
       available -= result.length;
-      // TODO(ricow): Remove when we track internal and pipe uses.
-      assert(resourceInfo != null || isPipe || isInternal);
-      if (resourceInfo != null) {
-        resourceInfo.totalRead += result.length;
-      }
+      totalRead += result.length;
     }
-    // TODO(ricow): Remove when we track internal and pipe uses.
-    assert(resourceInfo != null || isPipe || isInternal);
-    if (resourceInfo != null) {
-      resourceInfo.readCount++;
-      resourceInfo.lastRead = timestamp;
-    }
+    readCount++;
+    lastRead = timestamp;
     return result;
   }
 
   Datagram receive() {
     if (isClosing || isClosed) return null;
     var result = nativeRecvFrom();
     if (result is OSError) {
       reportError(result, "Receive failed");
       return null;
     }
     if (result != null) {
       // Read the next available. Available is only for the next datagram, not
       // the sum of all datagrams pending, so we need to call after each
       // receive. If available becomes > 0, the _NativeSocket will continue to
       // emit read events.
       available = nativeAvailable();
-      // TODO(ricow): Remove when we track internal and pipe uses.
-      assert(resourceInfo != null || isPipe || isInternal);
-      if (resourceInfo != null) {
-        resourceInfo.totalRead += result.data.length;
-      }
+      totalRead += result.data.length;
     }
-    // TODO(ricow): Remove when we track internal and pipe uses.
-    assert(resourceInfo != null || isPipe || isInternal);
-    if (resourceInfo != null) {
-      resourceInfo.readCount++;
-      resourceInfo.lastRead = timestamp;
-    }
+    readCount++;
+    lastRead = timestamp;
     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) {
       if (offset > buffer.length) {
         throw new RangeError.value(offset);
       }
@@ skipping 18 matching lines @@
       result = 0;
     }
     // The result may be negative, if we forced a short write for testing
     // purpose. In such case, don't mark writeAvailable as false, as we don't
     // know if we'll receive an event. It's better to just retry.
     if (result >= 0 && result < bytes) {
       writeAvailable = false;
     }
     // Negate the result, as stated above.
     if (result < 0) result = -result;
-    // TODO(ricow): Remove when we track internal and pipe uses.
-    assert(resourceInfo != null || isPipe || isInternal);
-    if (resourceInfo != null) {
-      resourceInfo.totalWritten += result;
-      resourceInfo.writeCount++;
-      resourceInfo.lastWrite = timestamp;
-    }
+    totalWritten += result;
+    writeCount++;
+    lastWrite = timestamp;
     return result;
   }
 
   int send(List<int> buffer, int offset, int bytes,
            InternetAddress address, int port) {
     if (isClosing || isClosed) return 0;
     _BufferAndStart bufferAndStart =
         _ensureFastAndSerializableByteData(
             buffer, offset, bytes);
     var result = nativeSendTo(
         bufferAndStart.buffer, bufferAndStart.start, bytes,
         address._in_addr, port);
     if (result is OSError) {
       scheduleMicrotask(() => reportError(result, "Send failed"));
       result = 0;
     }
-    // TODO(ricow): Remove when we track internal and pipe uses.
-    assert(resourceInfo != null || isPipe || isInternal);
-    if (resourceInfo != null) {
-      resourceInfo.totalWritten += result;
-      resourceInfo.writeCount++;
-      resourceInfo.lastWrite = timestamp;
-    }
+    totalWritten += result;
+    writeCount++;
+    lastWrite = timestamp;
     return result;
   }
 
   _NativeSocket accept() {
     // Don't issue accept if we're closing.
     if (isClosing || isClosed) return null;
     assert(available > 0);
     available--;
     tokens++;
     returnTokens(LISTENING_TOKEN_BATCH_SIZE);
     var socket = new _NativeSocket.normal();
     if (nativeAccept(socket) != true) return null;
     socket.localPort = localPort;
     socket.localAddress = address;
-    setupResourceInfo(socket);
-    // TODO(ricow): Remove when we track internal and pipe uses.
-    assert(resourceInfo != null || isPipe || isInternal);
-    if (resourceInfo != null) {
-      resourceInfo.totalRead += 1;
-      resourceInfo.lastRead = timestamp;
-    }
+    totalRead += 1;
+    lastRead = timestamp;
     return socket;
   }
 
   int get port {
     if (localPort != 0) return localPort;
     if (isClosing || isClosed) throw const SocketException.closed();
     var result = nativeGetPort();
     if (result is OSError) throw result;
     return localPort = result;
   }
@@ skipping 92 matching lines @@
             available = nativeAvailable();
             issueReadEvent();
             continue;
           }
         }
 
         var handler = eventHandlers[i];
         if (i == DESTROYED_EVENT) {
           assert(isClosing);
           assert(!isClosed);
-          // TODO(ricow): Remove/update when we track internal and pipe uses.
-          assert(resourceInfo != null || isPipe || isInternal);
-          if (resourceInfo != null) {
-            _SocketResourceInfo.SocketClosed(resourceInfo);
-          }
           isClosed = true;
           closeCompleter.complete();
           disconnectFromEventHandler();
           if (handler != null) handler();
           continue;
         }
 
         if (i == ERROR_EVENT) {
           if (!isClosing) {
             reportError(nativeGetError(), "");
@@ skipping 97 matching lines @@
     int fullData = (typeFlags & TYPE_TYPE_MASK) | data;
     assert(!isClosing);
     connectToEventHandler();
     _EventHandler._sendData(this, eventPort.sendPort, fullData);
   }
 
   void connectToEventHandler() {
     assert(!isClosed);
     if (eventPort == null) {
       eventPort = new RawReceivePort(multiplex);
-    }
-    if (!connectedResourceHandler) {
-      registerExtension('__getOpenSockets',
-                        _SocketResourceInfo.getOpenSockets);
-      registerExtension('__getSocketByID',
-                        _SocketResourceInfo.getSocketInfoMapByID);
-
-      connectedResourceHandler = true;
+      _sockets[_serviceId] = this;
     }
   }
 
   void disconnectFromEventHandler() {
     assert(eventPort != null);
     eventPort.close();
     eventPort = null;
+    _sockets.remove(_serviceId);
     // Now that we don't track this Socket anymore, we can clear the owner
     // field.
     owner = null;
   }
 
   // Check whether this is an error response from a native port call.
   static bool isErrorResponse(response) {
     return response is List && response[0] != _SUCCESS_RESPONSE;
   }
 
@@ skipping 84 matching lines @@
   void leaveMulticast(InternetAddress addr, NetworkInterface interface) {
     var interfaceAddr = multicastAddress(addr, interface);
     var interfaceIndex = interface == null ? 0 : interface.index;
     var result = nativeLeaveMulticast(
         addr._in_addr,
         interfaceAddr == null ? null : interfaceAddr._in_addr,
         interfaceIndex);
     if (result is OSError) throw result;
   }
 
+  String get _serviceTypePath => 'io/sockets';
+  String get _serviceTypeName => 'Socket';
+
+  String _JSONKind() {
+    return isListening ? "Listening" :
+           isPipe ? "Pipe" :
+           isInternal ? "Internal" : "Normal";
+  }
+
+  Map _toJSONPipe(bool ref) {
+    var name = 'Anonymous Pipe';
+    var r = {
+      'id': _servicePath,
+      'type': _serviceType(ref),
+      'name': name,
+      'user_name': name,
+      'kind': _JSONKind(),
+    };
+    if (ref) {
+      return r;
+    }
+    r['readClosed'] = isClosedRead;
+    r['writeClosed'] = isClosedWrite;
+    r['closing'] = isClosing;
+    r['fd'] = nativeGetSocketId();
+    if (owner != null) {
+      r['owner'] = owner._toJSON(true);
+    }
+    return r;
+  }
+
+  Map _toJSONInternal(bool ref) {
+    var name = 'Internal';
+    var r = {
+      'id': _servicePath,
+      'type': _serviceType(ref),
+      'name': name,
+      'user_name': name,
+      'kind': _JSONKind(),
+    };
+    if (ref) {
+      return r;
+    }
+    r['closing'] = isClosing;
+    r['fd'] = nativeGetSocketId();
+    if (owner != null) {
+      r['owner'] = owner._toJSON(true);
+    }
+    return r;
+  }
+
+  Map _toJSONNetwork(bool ref) {
+    var name = '${address.host}:$port';
+    if (isTcp && !isListening) name += " <-> ${remoteAddress.host}:$remotePort";
+    var r = {
+      'id': _servicePath,
+      'type': _serviceType(ref),
+      'name': name,
+      'user_name': name,
+      'kind': _JSONKind(),
+    };
+    if (ref) {
+      return r;
+    }
+    var protocol = isTcp ? "TCP" : isUdp ? "UDP" : null;
+    var localAddress;
+    var localPort;
+    var rAddress;
+    var rPort;
+    try {
+      localAddress = address.address;
+    } catch (e) { }
+    try {
+      localPort = port;
+    } catch (e) { }
+    try {
+      rAddress = this.remoteAddress.address;
+    } catch (e) { }
+    try {
+      rPort = remotePort;
+    } catch (e) { }
+    r['localAddress'] = localAddress;
+    r['localPort'] = localPort;
+    r['remoteAddress'] = rAddress;
+    r['remotePort'] = rPort;
+    r['protocol'] = protocol;
+    r['readClosed'] = isClosedRead;
+    r['writeClosed'] = isClosedWrite;
+    r['closing'] = isClosing;
+    r['listening'] = isListening;
+    r['fd'] = nativeGetSocketId();
+    if (owner != null) {
+      r['owner'] = owner._toJSON(true);
+    }
+    return r;
+  }
+
+  Map _toJSON(bool ref) {
+    var map;
+    if (isPipe) {
+      map =  _toJSONPipe(ref);
+    } else if (isInternal) {
+      map = _toJSONInternal(ref);
+    } else {
+      map = _toJSONNetwork(ref);
+    }
+    if (!ref) {
+      map['available'] = available;
+      map['totalRead'] = totalRead;
+      map['totalWritten'] = totalWritten;
+      map['readCount'] = totalWritten;
+      map['writeCount'] = writeCount;
+      map['lastRead'] = lastRead;
+      map['lastWrite'] = lastWrite;
+    }
+    return map;
+  }
 
   void nativeSetSocketId(int id) native "Socket_SetSocketId";
   nativeAvailable() native "Socket_Available";
   nativeRead(int len) native "Socket_Read";
   nativeRecvFrom() native "Socket_RecvFrom";
   nativeWrite(List<int> buffer, int offset, int bytes)
       native "Socket_WriteList";
   nativeSendTo(List<int> buffer, int offset, int bytes,
                List<int> address, int port)
       native "Socket_SendTo";
@@ skipping 132 matching lines @@
       _referencePort.listen((sendPort) {
         sendPort.send(
           [_socket.address,
            _socket.port,
            _v6Only]);
       });
     }
     return new _RawServerSocketReference(_referencePort.sendPort);
   }
 
+  Map _toJSON(bool ref) => _socket._toJSON(ref);
+
   void set _owner(owner) { _socket.owner = owner; }
 }
 
 
 class _RawServerSocketReference implements RawServerSocketReference {
   final SendPort _sendPort;
 
   _RawServerSocketReference(this._sendPort);
 
   Future<RawServerSocket> create() {
@@ skipping 157 matching lines @@
   }
 
   void _onSubscriptionStateChange() {
     if (_controller.hasListener) {
       _resume();
     } else {
       _socket.close();
     }
   }
 
+  Map _toJSON(bool ref) => _socket._toJSON(ref);
   void set _owner(owner) { _socket.owner = owner; }
 }
 
 
 patch class ServerSocket {
   /* patch */ static Future<ServerSocket> bind(address,
                                                int port,
                                                {int backlog: 0,
                                                 bool v6Only: false,
                                                 bool shared: false}) {
@@ skipping 42 matching lines @@
   int get port => _socket.port;
 
   InternetAddress get address => _socket.address;
 
   Future close() => _socket.close().then((_) => this);
 
   ServerSocketReference get reference {
     return new _ServerSocketReference(_socket.reference);
   }
 
+  Map _toJSON(bool ref) => _socket._toJSON(ref);
+
   void set _owner(owner) { _socket._owner = owner; }
 }
 
 
 patch class Socket {
   /* patch */ static Future<Socket> connect(host, int port, {sourceAddress}) {
     return RawSocket.connect(host, port, sourceAddress: sourceAddress).then(
         (socket) => new _Socket(socket));
   }
 }
@@ skipping 293 matching lines @@
     if (_detachReady != null) {
       _detachReady.complete(null);
     } else {
       if (_raw != null) {
         _raw.shutdown(SocketDirection.SEND);
         _disableWriteEvent();
       }
     }
   }
 
+  Map _toJSON(bool ref) => _raw._toJSON(ref);
   void set _owner(owner) { _raw._owner = owner; }
 }
 
 
 patch class RawDatagramSocket {
   /* patch */ static Future<RawDatagramSocket> bind(
       host, int port, {bool reuseAddress: true}) {
     return _RawDatagramSocket.bind(host, port, reuseAddress);
   }
 }
@@ skipping 135 matching lines @@
 Datagram _makeDatagram(List<int> data,
                        String address,
                        List<int> in_addr,
                        int port) {
   return new Datagram(
       data,
       new _InternetAddress(address, null, in_addr),
       port);
 }
 
+String _socketsStats() => _SocketsObservatory.toJSON();