fix: ws disconnection (#345)

* update ws removing custom implementation; fixed disconnection while connecting

* add test

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,
+  bool _manuallyDisconnected = false;
-      _connecting = false,
+  bool _connecting = false;
-      _reconnecting = 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<void> disconnect() async {
+    _connecting = false;
+    if (!_connectionCompleter.isCompleted) {
+      _connectionCompleter.complete();
+    }
     if (_manuallyDisconnected) {
       return;
     }

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

packages/stream_chat/test/src/api/web_socket_stub_test.dart

@@ -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<UnimplementedError>()));
-  });
-}

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<String> streamController =
+      final streamController = StreamController<String>.broadcast();
-          StreamController<String>.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<String> streamController =
+        StreamController<String>.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;