Add UDP support to dart:io

runtime/bin/socket_patch.dart

Index: runtime/bin/socket_patch.dart
diff --git a/runtime/bin/socket_patch.dart b/runtime/bin/socket_patch.dart
index 789dd001399664a3632d3f8e33467a49e198f5bb..0bb559eb2be4a2f89070db959586c411329ea757 100644
--- a/runtime/bin/socket_patch.dart
+++ b/runtime/bin/socket_patch.dart
@@ -125,6 +125,7 @@ class _InternetAddress implements InternetAddress {
 
   Future<InternetAddress> reverse() => _NativeSocket.reverseLookup(this);
 
+
   _InternetAddress(InternetAddressType this.type,
                    String this.address,
                    String this._host,
@@ -264,8 +265,8 @@ class _NativeSocket extends NativeFieldWrapperClass1 {
   // The type flags for this socket.
   final int typeFlags;
 
-  // Holds the port of the socket, null if not known.
-  int localPort;
+  // 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;
@@ -399,6 +400,12 @@ class _NativeSocket extends NativeFieldWrapperClass1 {
         });
   }
 
+  _NativeSocket.datagram(this.address, this.localPort)
+      : typeFlags = TYPE_NORMAL_SOCKET {
+    eventHandlers = new List(EVENT_COUNT + 1);
+    nativeCreateBindDatagram(address._sockaddr_storage, localPort);
+  }
+
   _NativeSocket.normal() : typeFlags = TYPE_NORMAL_SOCKET {
     eventHandlers = new List(EVENT_COUNT + 1);
   }
@@ -441,6 +448,16 @@ class _NativeSocket extends NativeFieldWrapperClass1 {
     return result;
   }
 
+  Datagram receive() {
+    if (isClosing || isClosed) return null;
+    var result = nativeRecvFrom();
+    if (result is OSError) {
+      reportError(result, "Receive 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;
@@ -471,6 +488,17 @@ class _NativeSocket extends NativeFieldWrapperClass1 {
     return result;
   }
 
+  int sendTo(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._sockaddr_storage, port);
+  }
+
   _NativeSocket accept() {
     // Don't issue accept if we're closing.
     if (isClosing || isClosed) return null;
@@ -482,7 +510,7 @@ class _NativeSocket extends NativeFieldWrapperClass1 {
   }
 
   int get port {
-    if (localPort != null) return localPort;
+    if (localPort != 0) return localPort;
     return localPort = nativeGetPort();
   }
 
@@ -693,26 +721,54 @@ class _NativeSocket extends NativeFieldWrapperClass1 {
     close();
   }
 
-  bool setOption(SocketOption option, bool enabled) {
+  bool getOption(SocketOption option) {
     if (option is! SocketOption) throw new ArgumentError(options);
-    if (enabled is! bool) throw new ArgumentError(enabled);
-    return nativeSetOption(option._value, enabled);
+    var result = nativeGetOption(option._value);
+    if (result is OSError) throw result;
+    return result;
+  }
+
+  bool setOption(SocketOption option, value) {
+    if (option is! SocketOption) throw new ArgumentError(options);
+    var result = nativeSetOption(option._value, value);
+    if (result is OSError) throw result;
+  }
+
+  void join(InternetAddress addr, NetworkInterface interface) {
+    var result = nativeJoinMulticast(addr._sockaddr_storage, 0);
+    if (result is OSError) throw result;
+  }
+
+  void leave(InternetAddress addr, NetworkInterface interface) {
+    var result = nativeLeaveMulticast(addr._sockaddr_storage, 0);
+    if (result is OSError) throw result;
   }
 
   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";
   nativeCreateConnect(List<int> addr,
                       int port) native "Socket_CreateConnect";
+  nativeCreateBindDatagram(List<int> addr,
+                           int port) native "Socket_CreateBindDatagram";
   nativeCreateBindListen(List<int> addr, int port, int backlog, bool v6Only)
       native "ServerSocket_CreateBindListen";
   nativeAccept(_NativeSocket socket) native "ServerSocket_Accept";
   int nativeGetPort() native "Socket_GetPort";
   List nativeGetRemotePeer() native "Socket_GetRemotePeer";
   OSError nativeGetError() native "Socket_GetError";
+  nativeGetOption(int option) native "Socket_GetOption";
   bool nativeSetOption(int option, bool enabled) native "Socket_SetOption";
+  bool nativeJoinMulticast(List<int> addr, int interface)
+      native "Socket_JoinMulticast";
+  bool nativeLeaveMulticast(List<int> addr, int interface)
+      native "Socket_LeaveMulticast";
 }
 
 
@@ -1275,3 +1331,152 @@ class _Socket extends Stream<List<int>> implements Socket {
     }
   }
 }
+
+
+class _RawDatagramSocket extends Stream implements RawDatagramSocket {
+  _NativeSocket _socket;
+  StreamController<RawSocketEvent> _controller;
+  bool _readEventsEnabled = true;
+  bool _writeEventsEnabled = true;
+
+  _RawDatagramSocket(InternetAddress address, int port) {
+    _socket = new _NativeSocket.datagram(address, port);
+    _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),
+      error: (e) {
+        _controller.addError(e);
+        close();
+      }
+    );
+  }
+
+  StreamSubscription<RawSocketEvent> listen(void onData(RawSocketEvent event),
+                                            {Function onError,
+                                             void onDone(),
+                                             bool cancelOnError}) {
+    return _controller.stream.listen(
+        onData,
+        onError: onError,
+        onDone: onDone,
+        cancelOnError: cancelOnError);
+  }
+
+  Future close() => _socket.close().then((_) => this);
+
+  bool send(Datagram datagram) {
+    var result = _socket.sendTo(datagram.data, 0, datagram.data.length,
+                                datagram.address, datagram.port);
+    if (result is OSError) {
+      throw result;
+    }
+    assert(result == datagram.data.length);
+    return true;
+  }
+
+  bool sendTo(List<data> buffer, InternetAddress address, int port) {
+    var result = _socket.sendTo(buffer, 0, buffer.length, address, port);
+    if (result is OSError) {
+      throw result;
+    }
+    assert(result == buffer.length);
+    return true;
+  }
+
+  Datagram receive() {
+    return _socket.receive();
+  }
+
+  void join(InternetAddress group, [NetworkInterface interface]) {
+    _socket.join(group, interface);
+  }
+
+  void leave(InternetAddress group, [NetworkInterface interface]) {
+    _socket.leave(group, interface);
+  }
+
+  bool get readEventsEnabled => _readEventsEnabled;
+  void set readEventsEnabled(bool value) {
+    if (value != _readEventsEnabled) {
+      _readEventsEnabled = value;
+      if (!_controller.isPaused) _resume();
+    }
+  }
+
+  bool get writeEventsEnabled => _writeEventsEnabled;
+  void set writeEventsEnabled(bool value) {
+    if (value != _writeEventsEnabled) {
+      _writeEventsEnabled = value;
+      if (!_controller.isPaused) _resume();
+    }
+  }
+
+  bool get multicastLoopback =>
+      _socket.getOption(SocketOption._IP_MULTICAST_LOOP);
+  void set multicastLoopback(bool value) =>
+      _socket.setOption(SocketOption._IP_MULTICAST_LOOP, value);
+
+  int get multicastTTL =>
+      _socket.getOption(SocketOption._IP_MULTICAST_TTL);
+  void set multicastTTL(int value) =>
+      _socket.setOption(SocketOption._IP_MULTICAST_TTL, value);
+
+  bool get broadcastEnabled =>
+      _socket.getOption(SocketOption._IP_BROADCAST);
+  void set broadcastEnabled(bool value) =>
+      _socket.setOption(SocketOption._IP_BROADCAST, value);
+
+  int get port => _socket.port;
+
+  InternetAddress get address => _socket.address;
+
+  _pause() {
+    _socket.setListening(read: false, write: false);
+  }
+
+  void _resume() {
+    _socket.setListening(read: _readEventsEnabled, write: _writeEventsEnabled);
+  }
+
+  void _onPauseStateChange() {
+    if (_controller.isPaused) {
+      _pause();
+    } else {
+      _resume();
+    }
+  }
+
+  void _onSubscriptionStateChange() {
+    if (_controller.hasListener) {
+      _resume();
+    } else {
+      close();
+    }
+  }
+}
+
+Datagram _makeDatagram(List<int> data,
+                       bool ipV6,
+                       String address,
+                       List<int> sockaddr_storage,
+                       int port) {
+  var addressType =
+      ipV6 ? InternetAddressType.IP_V6 : InternetAddressType.IP_V4;
+  return new Datagram(
+      data,
+      new _InternetAddress(addressType, address, null, sockaddr_storage),
+      port);
+}