From 6d259074eb62ef388bacbc25bd0250844001d4f9 Mon Sep 17 00:00:00 2001 From: Salvatore Giordano Date: Mon, 22 Mar 2021 09:41:52 +0100 Subject: [PATCH] fix: ws disconnection (#345) * update ws removing custom implementation; fixed disconnection while connecting * add test --- .../stream_chat/lib/src/api/websocket.dart | 22 +++++---- packages/stream_chat/pubspec.yaml | 2 +- .../test/src/api/web_socket_stub_test.dart | 9 ---- .../test/src/api/websocket_test.dart | 45 ++++++++++++++++--- 4 files changed, 55 insertions(+), 23 deletions(-) delete mode 100644 packages/stream_chat/test/src/api/web_socket_stub_test.dart diff --git a/packages/stream_chat/lib/src/api/websocket.dart b/packages/stream_chat/lib/src/api/websocket.dart index ea2bb291..0f78b1bf 100644 --- a/packages/stream_chat/lib/src/api/websocket.dart +++ b/packages/stream_chat/lib/src/api/websocket.dart @@ -6,9 +6,6 @@ import 'package:logging/logging.dart'; import 'package:meta/meta.dart'; import 'package:rxdart/rxdart.dart'; import 'package:stream_chat/src/api/connection_status.dart'; -import 'package:stream_chat/src/api/web_socket_channel_stub.dart' - if (dart.library.html) 'web_socket_channel_html.dart' - if (dart.library.io) 'web_socket_channel_io.dart'; import 'package:stream_chat/src/models/event.dart'; import 'package:stream_chat/src/models/user.dart'; import 'package:web_socket_channel/web_socket_channel.dart'; @@ -36,7 +33,7 @@ class WebSocket { this.connectPayload, this.handler, this.logger, - this.connectFunc = connectWebSocket, + this.connectFunc, this.reconnectionMonitorInterval = 1, this.healthCheckInterval = 20, this.reconnectionMonitorTimeout = 40, @@ -121,9 +118,9 @@ class WebSocket { WebSocketChannel _channel; Timer _healthCheck, _reconnectionMonitor; DateTime _lastEventAt; - bool _manuallyDisconnected = false, - _connecting = false, - _reconnecting = false; + bool _manuallyDisconnected = false; + bool _connecting = false; + bool _reconnecting = false; Event _decodeEvent(String source) => Event.fromJson(json.decode(source)); @@ -143,7 +140,8 @@ class WebSocket { logger.info('connecting to $_path'); - _channel = connectFunc(_path); + _channel = + connectFunc?.call(_path) ?? WebSocketChannel.connect(Uri.parse(_path)); _channel.stream.listen( (data) { final jsonData = json.decode(data); @@ -177,6 +175,10 @@ class WebSocket { } void _onData(data) { + if (_manuallyDisconnected) { + return; + } + final event = _decodeEvent(data); logger.info('received new event: $data'); @@ -302,6 +304,10 @@ class WebSocket { /// Disconnects the WS and releases eventual resources Future disconnect() async { + _connecting = false; + if (!_connectionCompleter.isCompleted) { + _connectionCompleter.complete(); + } if (_manuallyDisconnected) { return; } diff --git a/packages/stream_chat/pubspec.yaml b/packages/stream_chat/pubspec.yaml index 81aef23e..05023530 100644 --- a/packages/stream_chat/pubspec.yaml +++ b/packages/stream_chat/pubspec.yaml @@ -20,7 +20,7 @@ dependencies: mime: ^0.9.7 rxdart: ^0.25.0 uuid: ^2.2.2 - web_socket_channel: ^1.1.0 + web_socket_channel: ^1.2.0 dev_dependencies: build_runner: ^1.10.0 diff --git a/packages/stream_chat/test/src/api/web_socket_stub_test.dart b/packages/stream_chat/test/src/api/web_socket_stub_test.dart deleted file mode 100644 index e60706dc..00000000 --- a/packages/stream_chat/test/src/api/web_socket_stub_test.dart +++ /dev/null @@ -1,9 +0,0 @@ -import 'package:test/test.dart'; -import 'package:stream_chat/src/api/web_socket_channel_stub.dart'; - -void main() { - test('src/api/web_socket_stub_test', () { - expect( - () => connectWebSocket('fakeurl'), throwsA(isA())); - }); -} diff --git a/packages/stream_chat/test/src/api/websocket_test.dart b/packages/stream_chat/test/src/api/websocket_test.dart index d9486477..238cc850 100644 --- a/packages/stream_chat/test/src/api/websocket_test.dart +++ b/packages/stream_chat/test/src/api/websocket_test.dart @@ -1,6 +1,5 @@ import 'dart:async'; -import 'package:test/test.dart'; import 'package:logging/logging.dart'; import 'package:mockito/mockito.dart'; import 'package:stream_chat/src/api/connection_status.dart'; @@ -8,6 +7,7 @@ import 'package:stream_chat/src/api/websocket.dart'; import 'package:stream_chat/src/models/event.dart'; import 'package:stream_chat/src/models/user.dart'; import 'package:stream_chat/stream_chat.dart'; +import 'package:test/test.dart'; import 'package:web_socket_channel/web_socket_channel.dart'; class Functions { @@ -47,10 +47,9 @@ void main() { final mockWSChannel = MockWSChannel(); - final StreamController streamController = - StreamController.broadcast(); + final streamController = StreamController.broadcast(); - final computedUrl = + const computedUrl = 'wss://baseurl/connect?test=true&json=%7B%22payload%22%3A%22test%22%2C%22user_details%22%3A%7B%22id%22%3A%22testid%22%7D%7D'; when(connectFunc(computedUrl)).thenAnswer((_) => mockWSChannel); @@ -60,7 +59,7 @@ void main() { }); final timer = Timer.periodic( - Duration(milliseconds: 100), + const Duration(milliseconds: 100), (_) => streamController.sink.add('{}'), ); @@ -160,6 +159,42 @@ void main() { return connect; }); + test('should close correctly the controller while connecting', () async { + final handleFunc = MockFunctions().handleFunc; + + final ConnectWebSocket connectFunc = MockFunctions().connectFunc; + + final ws = WebSocket( + baseUrl: 'baseurl', + user: User(id: 'testid'), + logger: Logger('ws'), + connectParams: {'test': 'true'}, + connectPayload: {'payload': 'test'}, + handler: handleFunc, + connectFunc: connectFunc, + ); + + final mockWSChannel = MockWSChannel(); + + final StreamController streamController = + StreamController.broadcast(); + + final computedUrl = + 'wss://baseurl/connect?test=true&json=%7B%22payload%22%3A%22test%22%2C%22user_details%22%3A%7B%22id%22%3A%22testid%22%7D%7D'; + + when(connectFunc(computedUrl)).thenAnswer((_) => mockWSChannel); + when(mockWSChannel.sink).thenAnswer((_) => MockWSSink()); + when(mockWSChannel.stream).thenAnswer((_) { + return streamController.stream; + }); + + ws.connect(); + await ws.disconnect(); + streamController.add('{}'); + + verify(connectFunc(computedUrl)).called(1); + verifyNever(handleFunc(any)); + }); test('should run correctly health check', () async { final handleFunc = MockFunctions().handleFunc;