Wip fix online users status stream

Author: riccardocucia

example/lib/src/pages/auth.dart

@@ -5,7 +5,7 @@ import 'package:flutter_login/flutter_login.dart';
 import 'package:flutter_supabase_chat_core/flutter_supabase_chat_core.dart';
 import 'package:supabase_flutter/supabase_flutter.dart';
 
-import 'rooms.dart';
+import 'home.dart';
 
 class AuthScreen extends StatefulWidget {
   const AuthScreen({
@@ -83,7 +83,7 @@ class _AuthScreenState extends State<AuthScreen> {
         onSubmitAnimationCompleted: () {
           Navigator.of(context).pushReplacement(
             MaterialPageRoute(
-              builder: (context) => const RoomsPage(),
+              builder: (context) => const HomePage(),
             ),
           );
         },

lib/src/class/supabase_chat_core.dart

@@ -15,10 +15,25 @@ class SupabaseChatCore {
     Supabase.instance.client.auth.onAuthStateChange.listen((data) async {
       if (loggedSupabaseUser != null) {
         _loggedUser = await user(uid: loggedSupabaseUser!.id);
-        _currentUserOnlineStatusChannel =
-            getUserOnlineStatusChannel(loggedSupabaseUser!.id);
+        if (_currentUserOnlineStatusChannel == null) {
+          _currentUserOnlineStatusChannel ??=
+              _getUserOnlineStatusChannel(loggedSupabaseUser!.id);
+          _currentUserOnlineStatusChannel?.subscribe(
+            (status, error) async {
+              _userStatusSubscribed =
+                  status == RealtimeSubscribeStatus.subscribed;
+              if (_lastOnlineStatus == UserOnlineStatus.online) {
+                await _trackUserStatus();
+              } else {
+                await _currentUserOnlineStatusChannel?.untrack();
+              }
+            },
+          );
+        }
       } else {
         _loggedUser = null;
+        await _currentUserOnlineStatusChannel?.unsubscribe();
+        _userStatusSubscribed = false;
         _currentUserOnlineStatusChannel = null;
       }
     });
@@ -57,15 +72,14 @@ class SupabaseChatCore {
   /// Current logged in user. Is update automatically.
   types.User? get loggedUser => _loggedUser;
 
-  /// Returns user online status realtime channel .
-  RealtimeChannel getUserOnlineStatusChannel(String uid) =>
+  RealtimeChannel _getUserOnlineStatusChannel(String uid) =>
       client.channel('${config.realtimeOnlineUserPrefixChannel}$uid');
 
-  /// Returns a current user online status realtime channel .
+  UserOnlineStatus _lastOnlineStatus = UserOnlineStatus.offline;
+
   RealtimeChannel? _currentUserOnlineStatusChannel;
 
   bool _userStatusSubscribed = false;
-  bool _userStatusSubscribing = false;
 
   Future<void> _trackUserStatus() async {
     final userStatus = {
@@ -76,18 +90,7 @@ class SupabaseChatCore {
   }
 
   Future<void> setPresenceStatus(UserOnlineStatus status) async {
-    if (!_userStatusSubscribed && !_userStatusSubscribing) {
-      _userStatusSubscribing = true;
-      _currentUserOnlineStatusChannel?.subscribe(
-        (status, error) async {
-          if (status != RealtimeSubscribeStatus.subscribed) return;
-          _userStatusSubscribed = true;
-          _userStatusSubscribing = false;
-          await _trackUserStatus();
-        },
-      );
-    }
-
+    _lastOnlineStatus = status;
     switch (status) {
       case UserOnlineStatus.online:
         if (_userStatusSubscribed) {
@@ -102,6 +105,27 @@ class SupabaseChatCore {
     }
   }
 
+  final Map<String, RealtimeChannel> _onlineUserChannels = {};
+
+  UserOnlineStatus _userStatus(List<Presence> presences, String uid) =>
+      presences.map((e) => e.payload['uid']).contains(uid)
+          ? UserOnlineStatus.online
+          : UserOnlineStatus.offline;
+
+  /// Returns a stream of online user state from Supabase Realtime.
+  Stream<UserOnlineStatus> userOnlineStatus(String uid) {
+    final controller = StreamController<UserOnlineStatus>();
+    if(_onlineUserChannels[uid]==null) {
+      _onlineUserChannels[uid] = _getUserOnlineStatusChannel(uid);
+      _onlineUserChannels[uid]!.onPresenceJoin((payload) {
+        controller.sink.add(_userStatus(payload.newPresences, uid));
+      }).onPresenceLeave((payload) {
+        controller.sink.add(_userStatus(payload.currentPresences, uid));
+      }).subscribe();
+    }
+    return controller.stream;
+  }
+
   /// Returns the URL of an asset path in the bucket
   String getAssetUrl(String path) =>
       '${client.storage.url}/object/authenticated/${config.chatAssetsBucket}/$path';
@@ -324,21 +348,6 @@ class SupabaseChatCore {
         );
   }
 
-  /// Returns a stream of online user state from Supabase Realtime.
-  Stream<UserOnlineStatus> userOnlineStatus(String uid) {
-    final controller = StreamController<UserOnlineStatus>();
-    UserOnlineStatus userStatus(List<Presence> presences, String uid) =>
-        presences.map((e) => e.payload['uid']).contains(uid)
-            ? UserOnlineStatus.online
-            : UserOnlineStatus.offline;
-    getUserOnlineStatusChannel(uid).onPresenceJoin((payload) {
-      controller.sink.add(userStatus(payload.newPresences, uid));
-    }).onPresenceLeave((payload) {
-      controller.sink.add(userStatus(payload.currentPresences, uid));
-    }).subscribe();
-    return controller.stream;
-  }
-
   /// Returns a paginated list of rooms from Supabase. Only rooms where current
   /// logged in user exist are returned.
   Future<List<types.Room>> rooms({