blob: 1515403a76c5972eb6a0c07e7f6b3fcc943739a9 [file] [log] [blame]
// 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 = "127.0.0.1",
int port = 0,
int backlog = 0]) {
return _RawServerSocket.bind(address, port, backlog);
}
}
patch class RawSocket {
/* patch */ static Future<RawSocket> connect(host, int port) {
return _RawSocket.connect(host, port);
}
}
patch class InternetAddress {
/* patch */ static InternetAddress get LOOPBACK_IP_V4 {
return _InternetAddress.LOOPBACK_IP_V4;
}
/* patch */ static InternetAddress get LOOPBACK_IP_V6 {
return _InternetAddress.LOOPBACK_IP_V6;
}
/* patch */ static InternetAddress get ANY_IP_V4 {
return _InternetAddress.ANY_IP_V4;
}
/* patch */ static InternetAddress get ANY_IP_V6 {
return _InternetAddress.ANY_IP_V6;
}
/* patch */ static Future<List<InternetAddress>> lookup(
String host, {InternetAddressType type: InternetAddressType.IP_V4}) {
return _NativeSocket.lookup(host, type: type);
}
}
class _InternetAddress implements InternetAddress {
static const int _ADDRESS_LOOPBACK_IP_V4 = 0;
static const int _ADDRESS_LOOPBACK_IP_V6 = 1;
static const int _ADDRESS_ANY_IP_V4 = 2;
static const int _ADDRESS_ANY_IP_V6 = 3;
static _InternetAddress LOOPBACK_IP_V4 =
new _InternetAddress.fixed(_ADDRESS_LOOPBACK_IP_V4);
static _InternetAddress LOOPBACK_IP_V6 =
new _InternetAddress.fixed(_ADDRESS_LOOPBACK_IP_V6);
static _InternetAddress ANY_IP_V4 =
new _InternetAddress.fixed(_ADDRESS_ANY_IP_V4);
static _InternetAddress ANY_IP_V6 =
new _InternetAddress.fixed(_ADDRESS_ANY_IP_V6);
final InternetAddressType type;
final String address;
final String host;
final Uint8List _sockaddr_storage;
_InternetAddress(InternetAddressType this.type,
String this.address,
String this.host,
List<int> this._sockaddr_storage);
factory _InternetAddress.fixed(int id) {
var sockaddr = _fixed(id);
switch (id) {
case _ADDRESS_LOOPBACK_IP_V4:
return new _InternetAddress(
InternetAddressType.IP_V4, "127.0.0.1", "localhost", sockaddr);
case _ADDRESS_LOOPBACK_IP_V6:
return new _InternetAddress(
InternetAddressType.IP_V6, "::1", "ip6-localhost", sockaddr);
case _ADDRESS_ANY_IP_V4:
return new _InternetAddress(
InternetAddressType.IP_V4, "0.0.0.0", "0.0.0.0", sockaddr);
case _ADDRESS_ANY_IP_V6:
return new _InternetAddress(
InternetAddressType.IP_V6, "::", "::", sockaddr);
default:
assert(false);
throw new ArgumentError();
}
}
String toString() {
return "InternetAddress('$address', ${type.name})";
}
static Uint8List _fixed(int id) native "InternetAddress_Fixed";
}
// The _NativeSocket class encapsulates an OS socket.
class _NativeSocket extends NativeFieldWrapperClass1 {
// Bit flags used when communicating between the eventhandler and
// dart code. The EVENT flags are used to indicate events of
// interest when sending a message from dart code to the
// eventhandler. When receiving a message from the eventhandler the
// EVENT flags indicate the events that actually happened. The
// COMMAND flags are used to send commands from dart to the
// eventhandler. COMMAND flags are never received from the
// eventhandler. Additional flags are used to communicate other
// information.
static const int READ_EVENT = 0;
static const int WRITE_EVENT = 1;
static const int ERROR_EVENT = 2;
static const int CLOSED_EVENT = 3;
static const int FIRST_EVENT = READ_EVENT;
static const int LAST_EVENT = CLOSED_EVENT;
static const int EVENT_COUNT = LAST_EVENT - FIRST_EVENT + 1;
static const int CLOSE_COMMAND = 8;
static const int SHUTDOWN_READ_COMMAND = 9;
static const int SHUTDOWN_WRITE_COMMAND = 10;
static const int FIRST_COMMAND = CLOSE_COMMAND;
static const int LAST_COMMAND = SHUTDOWN_WRITE_COMMAND;
// Type flag send to the eventhandler providing additional
// information on the type of the file descriptor.
static const int LISTENING_SOCKET = 16;
static const int PIPE_SOCKET = 17;
static const int TYPE_NORMAL_SOCKET = 0;
static const int TYPE_LISTENING_SOCKET = 1 << LISTENING_SOCKET;
static const int TYPE_PIPE = 1 << PIPE_SOCKET;
// Native port messages.
static const HOST_NAME_LOOKUP = 0;
// Socket close state
bool isClosed = false;
bool isClosedRead = false;
bool isClosedWrite = false;
Completer closeCompleter = new Completer();
// Handlers and receive port for socket events from the event handler.
int eventMask = 0;
List eventHandlers;
ReceivePort eventPort;
// Indicates if native interrupts can be activated.
bool canActivateEvents = true;
// The type flags for this socket.
final int typeFlags;
// Holds the port of the socket, null if not known.
int localPort;
// Holds the address used to connect or bind the socket.
InternetAddress address;
// Native port for socket services.
static SendPort socketService;
static Future<List<InternetAddress>> lookup(
String host, {InternetAddressType type: InternetAddressType.IP_V4}) {
ensureSocketService();
return socketService.call([HOST_NAME_LOOKUP, host, type._value])
.then((response) {
if (isErrorResponse(response)) {
throw createError(response, "Failed host name lookup");
} else {
return response.skip(1).map((result) {
var type = new InternetAddressType._from(result[0]);
return new _InternetAddress(type, result[1], host, result[2]);
}).toList();
}
});
}
static Future<_NativeSocket> connect(host, int port) {
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 name lookup");
}
return list[0];
});
})
.then((address) {
ensureSocketService();
var socket = new _NativeSocket.normal();
socket.address = address;
var result = socket.nativeCreateConnect(
address._sockaddr_storage, port);
if (result is OSError) {
throw createError(result, "Connection failed");
} else {
var completer = new Completer();
// Setup handlers for receiving the first write event which
// indicate that the socket is fully connected.
socket.setHandlers(
write: () {
socket.setListening(read: false, write: false);
completer.complete(socket);
},
error: (e) {
socket.close();
completer.completeError(createError(e, "Connection failed"));
}
);
socket.setListening(read: false, write: true);
return completer.future;
}
});
}
static Future<_NativeSocket> bind(host,
int port,
int backlog) {
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 name lookup");
}
return list[0];
});
})
.then((address) {
var socket = new _NativeSocket.listen();
socket.address = address;
var result = socket.nativeCreateBindListen(address._sockaddr_storage,
port,
backlog);
if (result is OSError) {
throw new SocketIOException(
"Failed to create server socket", result);
}
if (port != 0) socket.localPort = port;
return socket;
});
}
_NativeSocket.normal() : typeFlags = TYPE_NORMAL_SOCKET {
eventHandlers = new List(EVENT_COUNT + 1);
_EventHandler._start();
}
_NativeSocket.listen() : typeFlags = TYPE_LISTENING_SOCKET {
eventHandlers = new List(EVENT_COUNT + 1);
_EventHandler._start();
}
_NativeSocket.pipe() : typeFlags = TYPE_PIPE {
eventHandlers = new List(EVENT_COUNT + 1);
_EventHandler._start();
}
int available() {
if (isClosed) return 0;
var result = nativeAvailable();
if (result is OSError) {
reportError(result, "Available failed");
return 0;
} else {
return result;
}
}
List<int> read(int len) {
if (len != null && len <= 0) {
throw new ArgumentError("Illegal length $len");
}
if (isClosed) return null;
var result = nativeRead(len == null ? -1 : len);
if (result is OSError) {
reportError(result, "Read 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;
if (bytes == null) bytes = buffer.length;
if (offset < 0) throw new RangeError.value(offset);
if (bytes < 0) throw new RangeError.value(bytes);
if ((offset + bytes) > buffer.length) {
throw new RangeError.value(offset + bytes);
}
if (offset is! int || bytes is! int) {
throw new ArgumentError("Invalid arguments to write on Socket");
}
if (isClosed) return 0;
if (bytes == 0) return 0;
_BufferAndStart bufferAndStart =
_ensureFastAndSerializableBuffer(buffer, offset, offset + bytes);
var result =
nativeWrite(bufferAndStart.buffer, bufferAndStart.start, bytes);
if (result is OSError) {
reportError(result, "Write failed");
result = 0;
}
return result;
}
_NativeSocket accept() {
var socket = new _NativeSocket.normal();
if (nativeAccept(socket) != true) return null;
socket.localPort = localPort;
socket.address = address;
return socket;
}
int get port {
if (localPort != null) return localPort;
return localPort = nativeGetPort();
}
int get remotePort {
return nativeGetRemotePeer()[1];
}
String get remoteHost {
return nativeGetRemotePeer()[0];
}
// Multiplexes socket events to the socket handlers.
void multiplex(int events) {
canActivateEvents = false;
for (int i = FIRST_EVENT; i <= LAST_EVENT; i++) {
if (((events & (1 << i)) != 0)) {
if (i == CLOSED_EVENT &&
typeFlags != TYPE_LISTENING_SOCKET &&
!isClosed) {
isClosedRead = true;
}
var handler = eventHandlers[i];
assert(handler != null);
if (i == WRITE_EVENT) {
// If the event was disabled before we had a chance to fire the event,
// discard it. If we register again, we'll get a new one.
if ((eventMask & (1 << i)) == 0) continue;
// Unregister the out handler before executing it. There is
// no need to notify the eventhandler as handlers are
// disabled while the event is handled.
eventMask &= ~(1 << i);
}
// Don't call the in handler if there is no data available
// after all.
if (i == READ_EVENT &&
typeFlags != TYPE_LISTENING_SOCKET &&
available() == 0) {
continue;
}
if (i == ERROR_EVENT) {
reportError(nativeGetError(), "");
} else if (!isClosed) {
handler();
}
}
}
if (isClosedRead && isClosedWrite) close();
canActivateEvents = true;
activateHandlers();
}
void setHandlers({read: null, write: null, error: null, closed: null}) {
eventHandlers[READ_EVENT] = read;
eventHandlers[WRITE_EVENT] = write;
eventHandlers[ERROR_EVENT] = error;
eventHandlers[CLOSED_EVENT] = closed;
}
void setListening({read: true, write: true}) {
eventMask = (1 << CLOSED_EVENT) | (1 << ERROR_EVENT);
if (read) eventMask |= (1 << READ_EVENT);
if (write) eventMask |= (1 << WRITE_EVENT);
activateHandlers();
}
Future get closeFuture => closeCompleter.future;
void activateHandlers() {
if (canActivateEvents && !isClosed) {
// If we don't listen for either read or write, disconnect as we won't
// get close and error events anyway.
if ((eventMask & ((1 << READ_EVENT) | (1 << WRITE_EVENT))) == 0) {
if (eventPort != null) disconnectFromEventHandler();
} else {
int data = eventMask;
data |= typeFlags;
if (isClosedRead) data &= ~(1 << READ_EVENT);
if (isClosedWrite) data &= ~(1 << WRITE_EVENT);
sendToEventHandler(data);
}
}
}
void close() {
if (!isClosed) {
sendToEventHandler(1 << CLOSE_COMMAND);
isClosed = true;
closeCompleter.complete(this);
}
// Outside the if support closing sockets created but never
// assigned any actual socket.
disconnectFromEventHandler();
}
void shutdown(SocketDirection direction) {
if (!isClosed) {
switch (direction) {
case SocketDirection.RECEIVE:
shutdownRead();
break;
case SocketDirection.SEND:
shutdownWrite();
break;
case SocketDirection.BOTH:
close();
break;
default:
throw new ArgumentError(direction);
}
}
}
void shutdownWrite() {
if (!isClosed) {
if (isClosedRead) {
close();
} else {
sendToEventHandler(1 << SHUTDOWN_WRITE_COMMAND);
}
isClosedWrite = true;
}
}
void shutdownRead() {
if (!isClosed) {
if (isClosedWrite) {
close();
} else {
sendToEventHandler(1 << SHUTDOWN_READ_COMMAND);
}
isClosedRead = true;
}
}
void sendToEventHandler(int data) {
connectToEventHandler();
assert(!isClosed);
_EventHandler._sendData(this, eventPort, data);
}
void connectToEventHandler() {
if (eventPort == null) {
eventPort = new ReceivePort();
eventPort.receive ((var message, _) => multiplex(message));
}
}
void disconnectFromEventHandler() {
if (eventPort != null) {
eventPort.close();
eventPort = null;
}
}
static void ensureSocketService() {
if (socketService == null) {
socketService = _NativeSocket.newServicePort();
}
}
// Check whether this is an error response from a native port call.
static bool isErrorResponse(response) {
return response is List && response[0] != _SUCCESS_RESPONSE;
}
// Create the appropriate error/exception from different returned
// error objects.
static createError(error, String message) {
if (error is OSError) {
return new SocketIOException(message, error);
} else if (error is List) {
assert(isErrorResponse(error));
switch (error[0]) {
case _ILLEGAL_ARGUMENT_RESPONSE:
return new ArgumentError();
case _OSERROR_RESPONSE:
return new SocketIOException(
message, new OSError(error[2], error[1]));
default:
return new Exception("Unknown error");
}
} else {
return new SocketIOException(message);
}
}
void reportError(error, String message) {
var e = createError(error, message);
// Invoke the error handler if any.
if (eventHandlers[ERROR_EVENT] != null) {
eventHandlers[ERROR_EVENT](e);
}
// For all errors we close the socket
close();
}
bool setOption(SocketOption option, bool enabled) {
if (option is! SocketOption) throw new ArgumentError(options);
if (enabled is! bool) throw new ArgumentError(enabled);
return nativeSetOption(option._value, enabled);
}
nativeAvailable() native "Socket_Available";
nativeRead(int len) native "Socket_Read";
nativeWrite(List<int> buffer, int offset, int bytes)
native "Socket_WriteList";
nativeCreateConnect(List<int> addr,
int port) native "Socket_CreateConnect";
nativeCreateBindListen(List<int> addr, int port, int backlog)
native "ServerSocket_CreateBindListen";
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";
static SendPort newServicePort() native "Socket_NewServicePort";
}
class _RawServerSocket extends Stream<RawSocket>
implements RawServerSocket {
final _NativeSocket _socket;
StreamController<RawSocket> _controller;
static Future<_RawServerSocket> bind(address,
int port,
int backlog) {
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)
.then((socket) => new _RawServerSocket(socket));
}
_RawServerSocket(this._socket) {
_controller = new StreamController(
onListen: _onSubscriptionStateChange,
onCancel: _onSubscriptionStateChange,
onPause: _onPauseStateChange,
onResume: _onPauseStateChange);
_socket.closeFuture.then((_) => _controller.close());
_socket.setHandlers(
read: () {
var socket = _socket.accept();
if (socket != null) _controller.add(new _RawSocket(socket));
},
error: (e) {
_controller.addError(e);
_controller.close();
}
);
}
StreamSubscription<RawSocket> listen(void onData(RawSocket event),
{void onError(Object error),
void onDone(),
bool cancelOnError}) {
return _controller.stream.listen(
onData,
onError: onError,
onDone: onDone,
cancelOnError: cancelOnError);
}
int get port => _socket.port;
void close() => _socket.close();
void _pause() {
_socket.setListening(read: false, write: false);
}
void _resume() {
_socket.setListening(read: true, write: false);
}
void _onSubscriptionStateChange() {
if (_controller.hasListener) {
_resume();
} else {
close();
}
}
void _onPauseStateChange() {
if (_controller.isPaused) {
_pause();
} else {
_resume();
}
}
}
class _RawSocket extends Stream<RawSocketEvent>
implements RawSocket {
final _NativeSocket _socket;
StreamController<RawSocketEvent> _controller;
bool _readEventsEnabled = true;
bool _writeEventsEnabled = true;
static Future<RawSocket> connect(host, int port) {
return _NativeSocket.connect(host, port)
.then((socket) => new _RawSocket(socket));
}
_RawSocket(this._socket) {
_controller = new StreamController(
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),
error: (e) {
_controller.addError(e);
close();
}
);
}
factory _RawSocket._writePipe(int fd) {
var native = new _NativeSocket.pipe();
native.isClosedRead = true;
if (fd != null) _getStdioHandle(native, fd);
return new _RawSocket(native);
}
factory _RawSocket._readPipe(int fd) {
var native = new _NativeSocket.pipe();
native.isClosedWrite = true;
if (fd != null) _getStdioHandle(native, fd);
return new _RawSocket(native);
}
StreamSubscription<RawSocketEvent> listen(void onData(RawSocketEvent event),
{void onError(Object error),
void onDone(),
bool cancelOnError}) {
return _controller.stream.listen(
onData,
onError: onError,
onDone: onDone,
cancelOnError: cancelOnError);
}
int available() => _socket.available();
List<int> read([int len]) => _socket.read(len);
int write(List<int> buffer, [int offset, int count]) =>
_socket.write(buffer, offset, count);
void close() => _socket.close();
void shutdown(SocketDirection direction) => _socket.shutdown(direction);
int get port => _socket.port;
int get remotePort => _socket.remotePort;
InternetAddress get address => _socket.address;
String get remoteHost => _socket.remoteHost;
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 setOption(SocketOption option, bool enabled) =>
_socket.setOption(option, enabled);
_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();
}
}
}
patch class ServerSocket {
/* patch */ static Future<ServerSocket> bind([address = "127.0.0.1",
int port = 0,
int backlog = 0]) {
return _ServerSocket.bind(address, port, backlog);
}
}
class _ServerSocket extends Stream<Socket>
implements ServerSocket {
final _socket;
static Future<_ServerSocket> bind(address,
int port,
int backlog) {
return _RawServerSocket.bind(address, port, backlog)
.then((socket) => new _ServerSocket(socket));
}
_ServerSocket(this._socket);
StreamSubscription<Socket> listen(void onData(Socket event),
{void onError(error),
void onDone(),
bool cancelOnError}) {
return _socket.map((rawSocket) => new _Socket(rawSocket)).listen(
onData,
onError: onError,
onDone: onDone,
cancelOnError: cancelOnError);
}
int get port => _socket.port;
void close() => _socket.close();
}
patch class Socket {
/* patch */ static Future<Socket> connect(host, int port) {
return RawSocket.connect(host, port).then(
(socket) => new _Socket(socket));
}
}
patch class SecureSocket {
/* patch */ factory SecureSocket._(RawSecureSocket rawSocket) =>
new _SecureSocket(rawSocket);
}
class _SocketStreamConsumer extends StreamConsumer<List<int>> {
StreamSubscription subscription;
final _Socket socket;
int offset;
List<int> buffer;
bool paused = false;
Completer streamCompleter;
_SocketStreamConsumer(this.socket);
Future<Socket> addStream(Stream<List<int>> stream) {
socket._ensureRawSocketSubscription();
streamCompleter = new Completer<Socket>();
if (socket._raw != null) {
subscription = stream.listen(
(data) {
assert(!paused);
assert(buffer == null);
buffer = data;
offset = 0;
write();
},
onError: (error) {
socket._consumerDone();
done(error);
},
onDone: () {
done();
},
cancelOnError: true);
}
return streamCompleter.future;
}
Future<Socket> close() {
socket._consumerDone();
return new Future.value(socket);
}
void write() {
try {
if (subscription == null) return;
assert(buffer != null);
// Write as much as possible.
offset += socket._write(buffer, offset, buffer.length - offset);
if (offset < buffer.length) {
if (!paused) {
paused = true;
// TODO(ajohnsen): It would be nice to avoid this check.
// Some info: socket._write can emit an event, if it fails to write.
// If the user closes the socket in that event, stop() will be called
// before we get a change to pause.
if (subscription == null) return;
subscription.pause();
}
socket._enableWriteEvent();
} else {
buffer = null;
if (paused) {
paused = false;
subscription.resume();
}
}
} catch (e) {
stop();
socket._consumerDone();
done(e);
}
}
void done([error]) {
if (streamCompleter != null) {
var tmp = streamCompleter;
streamCompleter = null;
if (error != null) {
tmp.completeError(error);
} else {
tmp.complete(socket);
}
}
}
void stop() {
if (subscription == null) return;
subscription.cancel();
subscription = null;
socket._disableWriteEvent();
}
}
class _Socket extends Stream<List<int>> implements Socket {
RawSocket _raw; // Set to null when the raw socket is closed.
bool _closed = false; // Set to true when the raw socket is closed.
StreamController _controller;
bool _controllerClosed = false;
_SocketStreamConsumer _consumer;
IOSink _sink;
var _subscription;
var _detachReady;
_Socket(RawSocket this._raw) {
_controller = new StreamController<List<int>>(
onListen: _onSubscriptionStateChange,
onCancel: _onSubscriptionStateChange,
onPause: _onPauseStateChange,
onResume: _onPauseStateChange);
_consumer = new _SocketStreamConsumer(this);
_sink = new IOSink(_consumer);
// Disable read events until there is a subscription.
_raw.readEventsEnabled = false;
// Disable write events until the consumer needs it for pending writes.
_raw.writeEventsEnabled = false;
}
factory _Socket._writePipe([int fd]) {
return new _Socket(new _RawSocket._writePipe(fd));
}
factory _Socket._readPipe([int fd]) {
return new _Socket(new _RawSocket._readPipe(fd));
}
_NativeSocket get _nativeSocket => _raw._socket;
StreamSubscription<List<int>> listen(void onData(List<int> event),
{void onError(error),
void onDone(),
bool cancelOnError}) {
return _controller.stream.listen(
onData,
onError: onError,
onDone: onDone,
cancelOnError: cancelOnError);
}
Encoding get encoding => _sink.encoding;
void set encoding(Encoding value) {
_sink.encoding = value;
}
void write(Object obj) => _sink.write(obj);
void writeln([Object obj = ""]) => _sink.writeln(obj);
void writeCharCode(int charCode) => _sink.writeCharCode(charCode);
void writeAll(Iterable objects, [sep = ""]) => _sink.writeAll(objects, sep);
void add(List<int> bytes) => _sink.add(bytes);
Future<Socket> addStream(Stream<List<int>> stream) {
return _sink.addStream(stream);
}
Future<Socket> close() => _sink.close();
Future<Socket> get done => _sink.done;
void destroy() {
// Destroy can always be called to get rid of a socket.
if (_raw == null) return;
_consumer.stop();
_closeRawSocket();
_controllerClosed = true;
_controller.close();
}
bool setOption(SocketOption option, bool enabled) {
if (_raw == null) return false;
return _raw.setOption(option, enabled);
}
int get port => _raw.port;
String get remoteHost => _raw.remoteHost;
int get remotePort => _raw.remotePort;
Future _detachRaw() {
_detachReady = new Completer();
_sink.close();
return _detachReady.future.then((_) {
assert(_consumer.buffer == null);
var raw = _raw;
_raw = null;
return [raw, _subscription];
});
}
// Ensure a subscription on the raw socket. Both the stream and the
// consumer needs a subscription as they share the error and done
// events from the raw socket.
void _ensureRawSocketSubscription() {
if (_subscription == null && _raw != null) {
_subscription = _raw.listen(_onData,
onError: _onError,
onDone: _onDone,
cancelOnError: true);
}
}
_closeRawSocket() {
var tmp = _raw;
_raw = null;
_closed = true;
tmp.close();
}
void _onSubscriptionStateChange() {
if (_controller.hasListener) {
_ensureRawSocketSubscription();
// Enable read events for providing data to subscription.
if (_raw != null) {
_raw.readEventsEnabled = true;
}
} else {
_controllerClosed = true;
if (_raw != null) {
_raw.shutdown(SocketDirection.RECEIVE);
}
}
}
void _onPauseStateChange() {
if (_raw != null) {
_raw.readEventsEnabled = !_controller.isPaused;
}
}
void _onData(event) {
switch (event) {
case RawSocketEvent.READ:
var buffer = _raw.read();
if (buffer != null) _controller.add(buffer);
break;
case RawSocketEvent.WRITE:
_consumer.write();
break;
case RawSocketEvent.READ_CLOSED:
_controllerClosed = true;
_controller.close();
break;
}
}
void _onDone() {
if (!_controllerClosed) {
_controllerClosed = true;
_controller.close();
}
_consumer.done();
}
void _onError(error) {
if (!_controllerClosed) {
_controllerClosed = true;
_controller.addError(error);
_controller.close();
}
_consumer.done(error);
}
int _write(List<int> data, int offset, int length) =>
_raw.write(data, offset, length);
void _enableWriteEvent() {
_raw.writeEventsEnabled = true;
}
void _disableWriteEvent() {
if (_raw != null) {
_raw.writeEventsEnabled = false;
}
}
void _consumerDone() {
if (_detachReady != null) {
_detachReady.complete(null);
} else {
if (_raw != null) {
_raw.shutdown(SocketDirection.SEND);
_disableWriteEvent();
}
}
}
}
class _SecureSocket extends _Socket implements SecureSocket {
_SecureSocket(RawSecureSocket raw) : super(raw);
void set onBadCertificate(bool callback(X509Certificate certificate)) {
if (_raw == null) {
throw new StateError("onBadCertificate called on destroyed SecureSocket");
}
_raw.onBadCertificate = callback;
}
X509Certificate get peerCertificate {
if (_raw == null) {
throw new StateError("peerCertificate called on destroyed SecureSocket");
}
return _raw.peerCertificate;
}
}