Add WebSocketChannel.`ready` field and IOWebSocketChannel factory `connectTimeout` parameter (dart-lang/web_socket_channel#240)
diff --git a/pkgs/web_socket_channel/CHANGELOG.md b/pkgs/web_socket_channel/CHANGELOG.md
index 1c5b128..6e0de0c 100644
--- a/pkgs/web_socket_channel/CHANGELOG.md
+++ b/pkgs/web_socket_channel/CHANGELOG.md
@@ -1,5 +1,9 @@
-## 2.2.1-dev
+## 2.3.0
+- Added a Future `ready` property to `WebSocketChannel`, which completes when
+ the connection is established
+- Added a `connectTimeout` parameter to the `IOWebSocketChannel.connect` factory,
+ which controls the timeout of the WebSocket Future.
- Use platform agnostic code in README example.
## 2.2.0
diff --git a/pkgs/web_socket_channel/lib/html.dart b/pkgs/web_socket_channel/lib/html.dart
index 4fe35a4..080b2bf 100644
--- a/pkgs/web_socket_channel/lib/html.dart
+++ b/pkgs/web_socket_channel/lib/html.dart
@@ -45,8 +45,15 @@
/// [_controller.local.stream].
String? _localCloseReason;
+ /// Completer for [ready].
+ late Completer<void> _readyCompleter;
+
+ @override
+ Future<void> get ready => _readyCompleter.future;
+
@override
Stream get stream => _controller.foreign.stream;
+
final _controller =
StreamChannelController(sync: true, allowForeignErrors: false);
@@ -71,12 +78,20 @@
/// Creates a channel wrapping [innerWebSocket].
HtmlWebSocketChannel(this.innerWebSocket) {
+ _readyCompleter = Completer();
if (innerWebSocket.readyState == WebSocket.OPEN) {
+ _readyCompleter.complete();
_listen();
} else {
+ if (innerWebSocket.readyState == WebSocket.CLOSING ||
+ innerWebSocket.readyState == WebSocket.CLOSED) {
+ _readyCompleter.completeError(WebSocketChannelException(
+ 'WebSocket state error: ${innerWebSocket.readyState}'));
+ }
// The socket API guarantees that only a single open event will be
// emitted.
innerWebSocket.onOpen.first.then((_) {
+ _readyCompleter.complete();
_listen();
});
}
@@ -84,8 +99,9 @@
// The socket API guarantees that only a single error event will be emitted,
// and that once it is no open or message events will be emitted.
innerWebSocket.onError.first.then((_) {
- _controller.local.sink
- .addError(WebSocketChannelException('WebSocket connection failed.'));
+ final error = WebSocketChannelException('WebSocket connection failed.');
+ _readyCompleter.completeError(error);
+ _controller.local.sink.addError(error);
_controller.local.sink.close();
});
diff --git a/pkgs/web_socket_channel/lib/io.dart b/pkgs/web_socket_channel/lib/io.dart
index a7bee8f..15f2f0d 100644
--- a/pkgs/web_socket_channel/lib/io.dart
+++ b/pkgs/web_socket_channel/lib/io.dart
@@ -32,9 +32,16 @@
@override
final Stream stream;
+
@override
final WebSocketSink sink;
+ /// Completer for [ready].
+ final Completer<void> _readyCompleter;
+
+ @override
+ Future<void> get ready => _readyCompleter.future;
+
/// The underlying [WebSocket], if this channel has connected.
///
/// If the future returned from [WebSocket.connect] has not yet completed, or
@@ -55,6 +62,10 @@
/// [pingInterval]. It defaults to `null`, indicating that ping messages are
/// disabled.
///
+ /// [connectTimeout] determines how long to wait for [WebSocket.connect]
+ /// before throwing a [TimeoutException]. If connectTimeout is null then the
+ /// connection process will never time-out.
+ ///
/// If there's an error connecting, the channel's stream emits a
/// [WebSocketChannelException] wrapping that error and then closes.
factory IOWebSocketChannel.connect(
@@ -62,21 +73,28 @@
Iterable<String>? protocols,
Map<String, dynamic>? headers,
Duration? pingInterval,
+ Duration? connectTimeout,
}) {
late IOWebSocketChannel channel;
final sinkCompleter = WebSocketSinkCompleter();
- final stream = StreamCompleter.fromFuture(
- WebSocket.connect(url.toString(), headers: headers, protocols: protocols)
- .then((webSocket) {
- webSocket.pingInterval = pingInterval;
- channel._webSocket = webSocket;
- sinkCompleter.setDestinationSink(_IOWebSocketSink(webSocket));
- return webSocket;
- }).catchError(
- (Object error) => throw WebSocketChannelException.from(error),
- ),
+ var future = WebSocket.connect(
+ url.toString(),
+ headers: headers,
+ protocols: protocols,
);
-
+ if (connectTimeout != null) {
+ future = future.timeout(connectTimeout);
+ }
+ final stream = StreamCompleter.fromFuture(future.then((webSocket) {
+ webSocket.pingInterval = pingInterval;
+ channel._webSocket = webSocket;
+ channel._readyCompleter.complete();
+ sinkCompleter.setDestinationSink(_IOWebSocketSink(webSocket));
+ return webSocket;
+ }).catchError((Object error, StackTrace stackTrace) {
+ channel._readyCompleter.completeError(error, stackTrace);
+ throw WebSocketChannelException.from(error);
+ }));
return channel =
IOWebSocketChannel._withoutSocket(stream, sinkCompleter.sink);
}
@@ -86,7 +104,8 @@
: _webSocket = socket,
stream = socket.handleError(
(error) => throw WebSocketChannelException.from(error)),
- sink = _IOWebSocketSink(socket);
+ sink = _IOWebSocketSink(socket),
+ _readyCompleter = Completer()..complete();
/// Creates a channel without a socket.
///
@@ -95,7 +114,8 @@
IOWebSocketChannel._withoutSocket(Stream stream, this.sink)
: _webSocket = null,
stream = stream.handleError(
- (error) => throw WebSocketChannelException.from(error));
+ (error) => throw WebSocketChannelException.from(error)),
+ _readyCompleter = Completer();
}
/// A [WebSocketSink] that forwards [close] calls to a `dart:io` [WebSocket].
diff --git a/pkgs/web_socket_channel/lib/src/channel.dart b/pkgs/web_socket_channel/lib/src/channel.dart
index fa0cb97..d912fcc 100644
--- a/pkgs/web_socket_channel/lib/src/channel.dart
+++ b/pkgs/web_socket_channel/lib/src/channel.dart
@@ -51,6 +51,10 @@
/// Before the connection has been closed, this will be `null`.
String? get closeReason => _webSocket.closeReason;
+ /// Future indicating if the connection has been established.
+ /// It completes on successful connection to the websocket.
+ final Future<void> ready = Future.value();
+
@override
Stream get stream => StreamView(_webSocket);
diff --git a/pkgs/web_socket_channel/pubspec.yaml b/pkgs/web_socket_channel/pubspec.yaml
index 4c83080..4041d87 100644
--- a/pkgs/web_socket_channel/pubspec.yaml
+++ b/pkgs/web_socket_channel/pubspec.yaml
@@ -1,5 +1,5 @@
name: web_socket_channel
-version: 2.2.1-dev
+version: 2.3.0
description: >-
StreamChannel wrappers for WebSockets. Provides a cross-platform
diff --git a/pkgs/web_socket_channel/test/html_test.dart b/pkgs/web_socket_channel/test/html_test.dart
index f7ec4d9..54b34d7 100644
--- a/pkgs/web_socket_channel/test/html_test.dart
+++ b/pkgs/web_socket_channel/test/html_test.dart
@@ -37,6 +37,9 @@
test('communicates using an existing WebSocket', () async {
final webSocket = WebSocket('ws://localhost:$port');
final channel = HtmlWebSocketChannel(webSocket);
+
+ expect(channel.ready, completes);
+
addTearDown(channel.sink.close);
final queue = StreamQueue(channel.stream);
@@ -59,6 +62,9 @@
await webSocket.onOpen.first;
final channel = HtmlWebSocketChannel(webSocket);
+
+ expect(channel.ready, completes);
+
addTearDown(channel.sink.close);
final queue = StreamQueue(channel.stream);
@@ -66,8 +72,29 @@
expect(await queue.next, equals('foo'));
});
+ test('communicates using an connecting WebSocket', () async {
+ final webSocket = WebSocket('ws://localhost:$port');
+
+ final channel = HtmlWebSocketChannel(webSocket);
+
+ expect(channel.ready, completes);
+
+ addTearDown(channel.sink.close);
+ });
+
+ test('communicates using an existing closed WebSocket', () async {
+ final webSocket = WebSocket('ws://localhost:$port');
+ webSocket.close();
+
+ final channel = HtmlWebSocketChannel(webSocket);
+ expect(channel.ready, throwsA(isA<WebSocketChannelException>()));
+ });
+
test('.connect defaults to binary lists', () async {
final channel = HtmlWebSocketChannel.connect('ws://localhost:$port');
+
+ expect(channel.ready, completes);
+
addTearDown(channel.sink.close);
final queue = StreamQueue(channel.stream);
@@ -81,6 +108,9 @@
test('.connect defaults to binary lists using platform independent api',
() async {
final channel = WebSocketChannel.connect(Uri.parse('ws://localhost:$port'));
+
+ expect(channel.ready, completes);
+
addTearDown(channel.sink.close);
final queue = StreamQueue(channel.stream);
@@ -94,6 +124,9 @@
test('.connect can use blobs', () async {
final channel = HtmlWebSocketChannel.connect('ws://localhost:$port',
binaryType: BinaryType.blob);
+
+ expect(channel.ready, completes);
+
addTearDown(channel.sink.close);
final queue = StreamQueue(channel.stream);
@@ -126,6 +159,7 @@
// invalid.
final channel = HtmlWebSocketChannel.connect(
'ws://localhost:${await serverChannel.stream.first}');
+ expect(channel.ready, throwsA(isA<WebSocketChannelException>()));
expect(channel.stream.toList(), throwsA(isA<WebSocketChannelException>()));
});
}
diff --git a/pkgs/web_socket_channel/test/io_test.dart b/pkgs/web_socket_channel/test/io_test.dart
index 111abb4..f681ac6 100644
--- a/pkgs/web_socket_channel/test/io_test.dart
+++ b/pkgs/web_socket_channel/test/io_test.dart
@@ -3,6 +3,7 @@
// BSD-style license that can be found in the LICENSE file.
@TestOn('vm')
+import 'dart:async';
import 'dart:io';
import 'package:test/test.dart';
@@ -28,6 +29,8 @@
final webSocket = await WebSocket.connect('ws://localhost:${server.port}');
final channel = IOWebSocketChannel(webSocket);
+ expect(channel.ready, completes);
+
var n = 0;
channel.stream.listen((message) {
if (n == 0) {
@@ -47,6 +50,7 @@
test('.connect communicates immediately', () async {
server = await HttpServer.bind('localhost', 0);
+ addTearDown(server.close);
server.transform(WebSocketTransformer()).listen((WebSocket webSocket) {
final channel = IOWebSocketChannel(webSocket);
channel.stream.listen((request) {
@@ -56,6 +60,9 @@
});
final channel = IOWebSocketChannel.connect('ws://localhost:${server.port}');
+
+ expect(channel.ready, completes);
+
channel.sink.add('ping');
channel.stream.listen(
@@ -69,6 +76,7 @@
test('.connect communicates immediately using platform independent api',
() async {
server = await HttpServer.bind('localhost', 0);
+ addTearDown(server.close);
server.transform(WebSocketTransformer()).listen((WebSocket webSocket) {
final channel = IOWebSocketChannel(webSocket);
channel.stream.listen((request) {
@@ -79,6 +87,9 @@
final channel =
WebSocketChannel.connect(Uri.parse('ws://localhost:${server.port}'));
+
+ expect(channel.ready, completes);
+
channel.sink.add('ping');
channel.stream.listen(
@@ -91,6 +102,7 @@
test('.connect with an immediate call to close', () async {
server = await HttpServer.bind('localhost', 0);
+ addTearDown(server.close);
server.transform(WebSocketTransformer()).listen((WebSocket webSocket) {
expect(() async {
final channel = IOWebSocketChannel(webSocket);
@@ -101,18 +113,23 @@
});
final channel = IOWebSocketChannel.connect('ws://localhost:${server.port}');
+
+ expect(channel.ready, completes);
+
await channel.sink.close(5678, 'raisin');
});
test('.connect wraps a connection error in WebSocketChannelException',
() async {
server = await HttpServer.bind('localhost', 0);
+ addTearDown(server.close);
server.listen((request) {
request.response.statusCode = 404;
request.response.close();
});
final channel = IOWebSocketChannel.connect('ws://localhost:${server.port}');
+ expect(channel.ready, throwsA(isA<WebSocketException>()));
expect(channel.stream.drain(), throwsA(isA<WebSocketChannelException>()));
});
@@ -122,6 +139,7 @@
String selector(List<String> receivedProtocols) => passedProtocol;
server = await HttpServer.bind('localhost', 0);
+ addTearDown(server.close);
server.listen((HttpRequest request) {
expect(
WebSocketTransformer.upgrade(request, protocolSelector: selector),
@@ -133,6 +151,7 @@
'ws://localhost:${server.port}',
protocols: [failedProtocol],
);
+ expect(channel.ready, throwsA(isA<WebSocketException>()));
expect(
channel.stream.drain(),
throwsA(isA<WebSocketChannelException>()),
@@ -144,6 +163,7 @@
String selector(List<String> receivedProtocols) => passedProtocol;
server = await HttpServer.bind('localhost', 0);
+ addTearDown(server.close);
server.listen((HttpRequest request) async {
final webSocket = await WebSocketTransformer.upgrade(
request,
@@ -155,7 +175,58 @@
final channel = IOWebSocketChannel.connect('ws://localhost:${server.port}',
protocols: [passedProtocol]);
+
+ expect(channel.ready, completes);
+
await channel.stream.drain();
expect(channel.protocol, passedProtocol);
});
+
+ test('.connects with a timeout parameters specified', () async {
+ server = await HttpServer.bind('localhost', 0);
+ addTearDown(server.close);
+ server.transform(WebSocketTransformer()).listen((webSocket) {
+ expect(() async {
+ final channel = IOWebSocketChannel(webSocket);
+ await channel.stream.drain();
+ expect(channel.closeCode, equals(5678));
+ expect(channel.closeReason, equals('raisin'));
+ }(), completes);
+ });
+
+ final channel = IOWebSocketChannel.connect(
+ 'ws://localhost:${server.port}',
+ connectTimeout: const Duration(milliseconds: 1000),
+ );
+ expect(channel.ready, completes);
+ await channel.sink.close(5678, 'raisin');
+ });
+
+ test('.respects timeout parameter when trying to connect', () async {
+ server = await HttpServer.bind('localhost', 0);
+ addTearDown(server.close);
+ server
+ .transform(StreamTransformer<HttpRequest, HttpRequest>.fromHandlers(
+ handleData: (data, sink) {
+ // Wait before we handle this request, to give the timeout a chance to
+ // kick in. We still want to make sure that we handle the request
+ // afterwards to not have false positives with the timeout
+ Timer(const Duration(milliseconds: 800), () {
+ sink.add(data);
+ });
+ }))
+ .transform(WebSocketTransformer())
+ .listen((webSocket) {
+ final channel = IOWebSocketChannel(webSocket);
+ channel.stream.drain();
+ });
+
+ final channel = IOWebSocketChannel.connect(
+ 'ws://localhost:${server.port}',
+ connectTimeout: const Duration(milliseconds: 500),
+ );
+
+ expect(channel.ready, throwsA(isA<TimeoutException>()));
+ expect(channel.stream.drain(), throwsA(isA<WebSocketChannelException>()));
+ });
}
