Index: runtime/bin/socket_patch.dart |
diff --git a/runtime/bin/socket_patch.dart b/runtime/bin/socket_patch.dart |
index 1480f1f97b41ef2331744cca1e9313d799dddfc4..1d256efe8cad5d681e266159ce39cbcd3b7a25c5 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); |
+ |
Anders Johnsen
2013/11/29 12:36:54
intentional?
Søren Gjesse
2013/12/12 11:38:46
No, reverted.
|
_InternetAddress(InternetAddressType this.type, |
String this.address, |
String this._host, |
@@ -246,6 +247,10 @@ class _NativeSocket extends NativeFieldWrapperClass1 { |
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; |
+ |
// Socket close state |
bool isClosed = false; |
bool isClosing = false; |
@@ -264,8 +269,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 +404,39 @@ class _NativeSocket extends NativeFieldWrapperClass1 { |
}); |
} |
+ 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._sockaddr_storage, 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; |
+ return socket; |
+ }); |
+ } |
+ |
+ _NativeSocket.datagram(this.address) |
+ : typeFlags = TYPE_NORMAL_SOCKET { |
+ eventHandlers = new List(EVENT_COUNT + 1); |
+ } |
+ |
_NativeSocket.normal() : typeFlags = TYPE_NORMAL_SOCKET { |
eventHandlers = new List(EVENT_COUNT + 1); |
} |
@@ -441,6 +479,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"); |
Anders Johnsen
2013/11/29 12:36:54
Put in scheduleMicrotask.
Søren Gjesse
2013/12/12 11:38:46
Done.
|
+ 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 +519,17 @@ class _NativeSocket extends NativeFieldWrapperClass1 { |
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( |
Anders Johnsen
2013/11/29 12:36:54
Use result?
Søren Gjesse
2013/12/12 11:38:46
Done.
|
+ 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 +541,7 @@ class _NativeSocket extends NativeFieldWrapperClass1 { |
} |
int get port { |
- if (localPort != null) return localPort; |
+ if (localPort != 0) return localPort; |
return localPort = nativeGetPort(); |
} |
@@ -697,26 +756,55 @@ class _NativeSocket extends NativeFieldWrapperClass1 { |
close(); |
} |
- bool setOption(SocketOption option, bool enabled) { |
+ bool getOption(SocketOption option) { |
Anders Johnsen
2013/11/29 12:36:54
I suppose the result is dynamic?
Søren Gjesse
2013/12/12 11:38:46
Done.
|
+ if (option is! SocketOption) throw new ArgumentError(options); |
+ var result = nativeGetOption(option._value, address.type._value); |
+ if (result is OSError) throw result; |
+ return result; |
+ } |
+ |
+ bool setOption(SocketOption option, value) { |
if (option is! SocketOption) throw new ArgumentError(options); |
- if (enabled is! bool) throw new ArgumentError(enabled); |
- return nativeSetOption(option._value, enabled); |
+ var result = nativeSetOption(option._value, address.type._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"; |
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"; |
OSError nativeGetError() native "Socket_GetError"; |
- bool nativeSetOption(int option, bool enabled) native "Socket_SetOption"; |
+ nativeGetOption(int option, int protocol) native "Socket_GetOption"; |
+ bool nativeSetOption(int option, int protocol, value) |
+ native "Socket_SetOption"; |
+ bool nativeJoinMulticast(List<int> addr, int interface) |
+ native "Socket_JoinMulticast"; |
+ bool nativeLeaveMulticast(List<int> addr, int interface) |
+ native "Socket_LeaveMulticast"; |
} |
@@ -1279,3 +1367,154 @@ 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(this._socket) { |
Anders Johnsen
2013/11/29 12:36:54
Update with zone wrappers, like in RawSocket.
Søren Gjesse
2013/12/12 11:38:46
Done.
|
+ _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(); |
+ } |
+ ); |
+ } |
+ |
+ static Future<RawDatagramSocket> bind( |
Anders Johnsen
2013/11/29 12:36:54
patch method?
Søren Gjesse
2013/12/12 11:38:46
Done.
|
+ host, int port, bool reuseAddress) { |
+ if (port < 0 || port > 0xffff) |
+ throw new ArgumentError("Invalid port $port"); |
+ return _NativeSocket.bindDatagram(host, port, reuseAddress) |
+ .then((socket) => new _RawDatagramSocket(socket)); |
+ } |
+ |
+ 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(List<data> buffer, InternetAddress address, int port) { |
+ var result = _socket.send(buffer, 0, buffer.length, address, port); |
+ if (result is OSError) { |
+ throw result; |
+ } |
+ assert(result == buffer.length); |
Anders Johnsen
2013/11/29 12:36:54
It can be 0?
Søren Gjesse
2013/12/12 11:38:46
Yes, Changed to just call the native socket as the
|
+ 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 multicastHops => |
+ _socket.getOption(SocketOption._IP_MULTICAST_HOPS); |
+ void set multicastHops(int value) => |
+ _socket.setOption(SocketOption._IP_MULTICAST_HOPS, value); |
+ |
+ NetworkInterface get multicastInterface => |
+ throw "Not implemented"; |
+ void set multicastInterface(NetworkInterface value) => |
+ throw "Not implemented"; |
+ |
+ 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); |
+} |