Merge branch 'develop' into cds-189

packages/stream_chat/lib/src/client/client.dart

@@ -25,12 +25,14 @@ import 'package:stream_chat/src/core/models/member.dart';
 import 'package:stream_chat/src/core/models/message.dart';
 import 'package:stream_chat/src/core/models/own_user.dart';
 import 'package:stream_chat/src/core/models/user.dart';
+import 'package:stream_chat/src/core/platform_detector/platform_detector.dart';
 import 'package:stream_chat/src/core/util/utils.dart';
 import 'package:stream_chat/src/db/chat_persistence_client.dart';
 import 'package:stream_chat/src/event_type.dart';
 import 'package:stream_chat/src/location.dart';
 import 'package:stream_chat/src/ws/connection_status.dart';
 import 'package:stream_chat/src/ws/websocket.dart';
+import 'package:stream_chat/version.dart';
 
 /// Handler function used for logging records. Function requires a single
 /// [LogRecord] as the only parameter.
@@ -42,6 +44,10 @@ final _levelEmojiMapper = {
   Level.SEVERE: '🚨',
 };
 
+final _userAgent = 'stream-chat-dart-client-'
+    '${CurrentPlatform.name}-'
+    '${PACKAGE_VERSION.split('+')[0]}';
+
 /// The official Dart client for Stream Chat,
 /// a service for building chat applications.
 /// This library can be used on any Dart project and on both mobile and web apps
@@ -80,6 +86,7 @@ class StreamChatClient {
       location: location,
       connectTimeout: connectTimeout,
       receiveTimeout: receiveTimeout,
+      headers: {'X-Stream-Client': _userAgent},
     );
 
     _chatApi = chatApi ??
@@ -99,6 +106,7 @@ class StreamChatClient {
           tokenManager: _tokenManager,
           handler: handleEvent,
           logger: detachedLogger('🔌'),
+          queryParameters: {'X-Stream-Client': _userAgent},
         );
 
     _retryPolicy = retryPolicy ??
@@ -308,23 +316,21 @@ class StreamChatClient {
     );
 
     final ownUser = OwnUser.fromUser(user);
-    state.user = ownUser;
+    state.currentUser = ownUser;
 
-    if (!connectWebSocket) {
-      return ownUser;
-    }
+    if (!connectWebSocket) return ownUser;
 
     try {
       if (_originalChatPersistenceClient != null) {
         _chatPersistenceClient = _originalChatPersistenceClient;
         await _chatPersistenceClient!.connect(ownUser.id);
       }
-      final res = await openConnection();
-      return res;
+      final connectedUser = await openConnection();
+      return state.currentUser = connectedUser;
     } catch (e, stk) {
       if (e is StreamWebSocketError && e.isRetriable) {
         final event = await _chatPersistenceClient?.getConnectionInfo();
-        if (event != null) return event.me?.merge(ownUser) ?? ownUser;
+        if (event != null) return ownUser.merge(event.me);
       }
       logger.severe('error connecting user : ${ownUser.id}', e, stk);
       rethrow;
@@ -334,12 +340,12 @@ class StreamChatClient {
   /// Creates a new WebSocket connection with the current user.
   Future<OwnUser> openConnection() async {
     assert(
-      state.user != null,
+      state.currentUser != null,
       'User is not set on client, '
       'use `connectUser` or `connectAnonymousUser` instead',
     );
 
-    final user = state.user!;
+    final user = state.currentUser!;
 
     logger.info('Opening web-socket connection for ${user.id}');
 
@@ -363,7 +369,7 @@ class StreamChatClient {
 
     try {
       final event = await _ws.connect(user);
-      return event.me?.merge(user) ?? user;
+      return user.merge(event.me);
     } catch (e, stk) {
       logger.severe('error connecting ws', e, stk);
       rethrow;
@@ -378,7 +384,7 @@ class StreamChatClient {
   void closeConnection() {
     if (wsConnectionStatus == ConnectionStatus.disconnected) return;
 
-    logger.info('Closing web-socket connection for ${state.user?.id}');
+    logger.info('Closing web-socket connection for ${state.currentUser?.id}');
     _wsConnectionStatus = ConnectionStatus.disconnected;
 
     _connectionStatusSubscription?.cancel();
@@ -388,9 +394,6 @@ class StreamChatClient {
   }
 
   void _handleHealthCheckEvent(Event event) {
-    final user = event.me;
-    if (user != null) state.user = user;
-
     final connectionId = event.connectionId;
     if (connectionId != null) {
       _connectionIdManager.setConnectionId(connectionId);
@@ -1312,7 +1315,7 @@ class StreamChatClient {
   /// If [flushChatPersistence] is true the client deletes all offline
   /// user's data.
   Future<void> disconnectUser({bool flushChatPersistence = false}) async {
-    logger.info('Disconnecting user : ${state.user?.id}');
+    logger.info('Disconnecting user : ${state.currentUser?.id}');
 
     // resetting state
     state.dispose();
@@ -1355,16 +1358,15 @@ class ClientState {
     _subscriptions.addAll([
       _client
           .on()
-          .where((event) => event.me != null)
-          .map((e) => e.me)
+          .where((event) =>
+              event.me != null && event.type != EventType.healthCheck)
+          .map((e) => e.me!)
           .listen((user) {
-        _userController.add(user);
-        final totalUnreadCount = user?.totalUnreadCount;
-        if (totalUnreadCount != null) {
-          _totalUnreadCountController.add(totalUnreadCount);
-        }
+        currentUser = currentUser?.merge(user) ?? user;
+        final totalUnreadCount = user.totalUnreadCount;
+        _totalUnreadCountController.add(totalUnreadCount);
 
-        final unreadChannels = user?.unreadChannels;
+        final unreadChannels = user.unreadChannels;
         if (unreadChannels != null) {
           _unreadChannelsController.add(unreadChannels);
         }
@@ -1408,8 +1410,8 @@ class ClientState {
 
   void _listenUserUpdated() {
     _subscriptions.add(_client.on(EventType.userUpdated).listen((event) {
-      if (event.user!.id == user!.id) {
-        user = OwnUser.fromJson(event.user!.toJson());
+      if (event.user!.id == currentUser!.id) {
+        currentUser = OwnUser.fromJson(event.user!.toJson());
       }
       updateUser(event.user);
     }));
@@ -1431,9 +1433,10 @@ class ClientState {
 
   final StreamChatClient _client;
 
-  /// Update user information
-  set user(OwnUser? user) {
-    _userController.add(user);
+  /// Sets the user currently interacting with the client
+  /// note: this fully overrides the [currentUser]
+  set currentUser(OwnUser? user) {
+    _currentUserController.add(user);
   }
 
   /// Update all the [users] with the provided [userList]
@@ -1450,10 +1453,24 @@ class ClientState {
   void updateUser(User? user) => updateUsers([user]);
 
   /// The current user
-  OwnUser? get user => _userController.valueOrNull;
+  OwnUser? get currentUser => _currentUserController.valueOrNull;
 
   /// The current user as a stream
-  Stream<OwnUser?> get userStream => _userController.stream;
+  Stream<OwnUser?> get currentUserStream => _currentUserController.stream;
+
+  // coverage:ignore-start
+
+  /// The current user
+  @Deprecated('Use `.currentUser` instead, Will be removed in future releases')
+  OwnUser? get user => _currentUserController.valueOrNull;
+
+  /// The current user as a stream
+  @Deprecated(
+    'Use `.currentUserStream` instead, Will be removed in future releases',
+  )
+  Stream<OwnUser?> get userStream => _currentUserController.stream;
+
+  // coverage:ignore-end
 
   /// The current user
   Map<String, User> get users => _usersController.value;
@@ -1485,7 +1502,7 @@ class ClientState {
   }
 
   final _channelsController = BehaviorSubject<Map<String, Channel>>.seeded({});
-  final _userController = BehaviorSubject<OwnUser?>();
+  final _currentUserController = BehaviorSubject<OwnUser?>();
   final _usersController = BehaviorSubject<Map<String, User>>.seeded({});
   final _unreadChannelsController = BehaviorSubject<int>.seeded(0);
   final _totalUnreadCountController = BehaviorSubject<int>.seeded(0);
@@ -1493,7 +1510,7 @@ class ClientState {
   /// Call this method to dispose this object
   void dispose() {
     _subscriptions.forEach((s) => s.cancel());
-    _userController.close();
+    _currentUserController.close();
     _unreadChannelsController.close();
     _totalUnreadCountController.close();
     channels.values.forEach((c) => c.dispose());