fix changelog

Author: lohanidamodar

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 12.0.2
+
+* Fixed realtime multiple subscription issues
+
 ## 12.0.1
 
 * Fixed parameters using enum types

README.md

@@ -21,7 +21,7 @@ Add this to your package's `pubspec.yaml` file:
 
 ```yml
 dependencies:
-  appwrite: ^12.0.2
+  appwrite: ^12.0.1
 ```
 
 You can install packages from the command line:

lib/services/account.dart

@@ -188,7 +188,7 @@ class Account extends Service {
     ///
     /// Add an authenticator app to be used as an MFA factor. Verify the
     /// authenticator using the [verify
-    /// authenticator](/docs/references/cloud/client-web/account#verifyAuthenticator)
+    /// authenticator](/docs/references/cloud/client-web/account#updateMfaAuthenticator)
     /// method.
     Future<models.MfaType> createMfaAuthenticator({required enums.AuthenticatorType type}) async {
         final String apiPath = '/account/mfa/authenticators/{type}'.replaceAll('{type}', type.value);
@@ -209,8 +209,8 @@ class Account extends Service {
     /// Verify Authenticator
     ///
     /// Verify an authenticator app after adding it using the [add
-    /// authenticator](/docs/references/cloud/client-web/account#addAuthenticator)
-    /// method.
+    /// authenticator](/docs/references/cloud/client-web/account#createMfaAuthenticator)
+    /// method. add 
     Future<models.User> updateMfaAuthenticator({required enums.AuthenticatorType type, required String otp}) async {
         final String apiPath = '/account/mfa/authenticators/{type}'.replaceAll('{type}', type.value);
 

lib/src/client_browser.dart

@@ -43,7 +43,7 @@ class ClientBrowser extends ClientBase with ClientMixin {
       'x-sdk-name': 'Flutter',
       'x-sdk-platform': 'client',
       'x-sdk-language': 'flutter',
-      'x-sdk-version': '12.0.2',
+      'x-sdk-version': '12.0.1',
       'X-Appwrite-Response-Format': '1.5.0',
     };
 

lib/src/client_io.dart

@@ -64,7 +64,7 @@ class ClientIO extends ClientBase with ClientMixin {
       'x-sdk-name': 'Flutter',
       'x-sdk-platform': 'client',
       'x-sdk-language': 'flutter',
-      'x-sdk-version': '12.0.2',
+      'x-sdk-version': '12.0.1',
       'X-Appwrite-Response-Format' : '1.5.0',
     };
 

lib/src/enums/flag.dart

@@ -142,6 +142,7 @@ enum Flag {
     palau(value: 'pw'),
     papuaNewGuinea(value: 'pg'),
     poland(value: 'pl'),
+    frenchPolynesia(value: 'pf'),
     northKorea(value: 'kp'),
     portugal(value: 'pt'),
     paraguay(value: 'py'),

lib/src/realtime_mixin.dart

@@ -2,7 +2,7 @@ import 'dart:async';
 import 'dart:convert';
 import 'package:flutter/foundation.dart';
 import 'package:web_socket_channel/web_socket_channel.dart';
-import 'package:web_socket_channel/status.dart' as status;
+import 'package:web_socket_channel/status.dart';
 import 'exception.dart';
 import 'realtime_subscription.dart';
 import 'client.dart';
@@ -15,46 +15,35 @@ typedef GetFallbackCookie = String? Function();
 
 mixin RealtimeMixin {
   late Client client;
-  final Set<String> _channels = {};
+  final Map<String, List<StreamController<RealtimeMessage>>> _channels = {};
   WebSocketChannel? _websok;
   String? _lastUrl;
   late WebSocketFactory getWebSocket;
   GetFallbackCookie? getFallbackCookie;
   int? get closeCode => _websok?.closeCode;
-  int _subscriptionsCounter = 0;
-  Map<int, RealtimeSubscription> _subscriptions = {};
-  bool _notifyDone = true;
-  StreamSubscription? _websocketSubscription;
-  bool _creatingSocket = false;
 
   Future<dynamic> _closeConnection() async {
-    await _websocketSubscription?.cancel();
-    await _websok?.sink.close(status.normalClosure, 'Ending session');
+    await _websok?.sink.close(normalClosure);
     _lastUrl = null;
   }
 
   _createSocket() async {
-    if(_creatingSocket || _channels.isEmpty) return;
-    _creatingSocket = true;
     final uri = _prepareUri();
     if (_websok == null) {
       _websok = await getWebSocket(uri);
       _lastUrl = uri.toString();
     } else {
       if (_lastUrl == uri.toString() && _websok?.closeCode == null) {
-        _creatingSocket = false;
         return;
       }
-      _notifyDone = false;
       await _closeConnection();
       _lastUrl = uri.toString();
       _websok = await getWebSocket(uri);
-      _notifyDone = true;
     }
     debugPrint('subscription: $_lastUrl');
 
     try {
-      _websocketSubscription = _websok?.stream.listen((response) {
+      _websok?.stream.listen((response) {
         final data = RealtimeResponse.fromJson(response);
         switch (data.type) {
           case 'error':
@@ -78,25 +67,28 @@ mixin RealtimeMixin {
             break;
           case 'event':
             final message = RealtimeMessage.fromMap(data.data);
-            for (var subscription in _subscriptions.values) {
-              for (var channel in message.channels) {
-                if (subscription.channels.contains(channel)) {
-                  subscription.controller.add(message);
+            for(var channel in message.channels) {
+              if (_channels[channel] != null) {
+                for( var stream in _channels[channel]!) {
+                  stream.sink.add(message);
                 }
               }
             }
             break;
         }
       }, onDone: () {
-        if (!_notifyDone || _creatingSocket) return;
-        for (var subscription in _subscriptions.values) {
-          subscription.close();
+        for (var list in _channels.values) {
+          for (var stream in list) {
+            stream.close();
+          }
         }
         _channels.clear();
         _closeConnection();
       }, onError: (err, stack) {
-        for (var subscription in _subscriptions.values) {
-          subscription.controller.addError(err, stack);
+        for (var list in _channels.values) {
+          for (var stream in list) {
+            stream.sink.addError(err, stack);
+          }
         }
         if (_websok?.closeCode != null && _websok?.closeCode != 1008) {
           debugPrint("Reconnecting in one second.");
@@ -111,8 +103,6 @@ mixin RealtimeMixin {
         throw AppwriteException(e.message);
       }
       throw AppwriteException(e.toString());
-    } finally {
-      _creatingSocket = false;
     }
   }
 
@@ -128,46 +118,40 @@ mixin RealtimeMixin {
       port: uri.port,
       queryParameters: {
         "project": client.config['project'],
-        "channels[]": _channels.toList(),
+        "channels[]": _channels.keys.toList(),
       },
       path: uri.path + "/realtime",
     );
   }
 
   RealtimeSubscription subscribeTo(List<String> channels) {
     StreamController<RealtimeMessage> controller = StreamController.broadcast();
-    _channels.addAll(channels);
+    for(var channel in channels) {
+      if (!_channels.containsKey(channel)) {
+        _channels[channel] = [];
+      }
+      _channels[channel]!.add(controller);
+    }
     Future.delayed(Duration.zero, () => _createSocket());
-    int id = DateTime.now().microsecondsSinceEpoch;
     RealtimeSubscription subscription = RealtimeSubscription(
-        controller: controller,
-        channels: channels,
+        stream: controller.stream,
         close: () async {
-          _subscriptions.remove(id);
-          _subscriptionsCounter--;
           controller.close();
-          _cleanup(channels);
-
-          if (_channels.isNotEmpty) {
+          for(var channel in channels) {
+            _channels[channel]!.remove(controller);
+            if (_channels[channel]!.isEmpty) {
+              _channels.remove(channel);
+            }
+          }
+          if(_channels.isNotEmpty) {
             await Future.delayed(Duration.zero, () => _createSocket());
           } else {
             await _closeConnection();
           }
         });
-    _subscriptions[id] = subscription;
     return subscription;
   }
 
-  void _cleanup(List<String> channels) {
-    for (var channel in channels) {
-      bool found = _subscriptions.values
-          .any((subscription) => subscription.channels.contains(channel));
-      if (!found) {
-        _channels.remove(channel);
-      }
-    }
-  }
-
   void handleError(RealtimeResponse response) {
     if (response.data['code'] == 1008) {
       throw AppwriteException(response.data["message"], response.data["code"]);

lib/src/realtime_subscription.dart

@@ -1,22 +1,13 @@
-import 'dart:async';
-
 import 'realtime_message.dart';
 
 /// Realtime Subscription
 class RealtimeSubscription {
   /// Stream of [RealtimeMessage]s
   final Stream<RealtimeMessage> stream;
 
-  final StreamController<RealtimeMessage> controller;
-
-  /// List of channels
-  List<String> channels;
-
   /// Closes the subscription
   final Future<void> Function() close;
 
   /// Initializes a [RealtimeSubscription]
-  RealtimeSubscription(
-      {required this.close, required this.channels, required this.controller})
-      : stream = controller.stream;
+  RealtimeSubscription({required this.stream, required this.close});
 }

pubspec.yaml

@@ -1,5 +1,5 @@
 name: appwrite
-version: 12.0.2
+version: 12.0.1
 description: Appwrite is an open-source self-hosted backend server that abstract and simplify complex and repetitive development tasks behind a very simple REST API
 homepage: https://appwrite.io
 repository: https://github.com/appwrite/sdk-for-flutter

test/src/realtime_subscription_test.dart

@@ -1,20 +1,20 @@
+import 'package:mockito/mockito.dart';
 import 'package:appwrite/src/realtime_message.dart';
 import 'package:appwrite/src/realtime_subscription.dart';
 import 'package:flutter_test/flutter_test.dart';
-import 'dart:async';
+
+class MockStream<T> extends Mock implements Stream<T> {}
+
+
 
 void main() {
   group('RealtimeSubscription', () {
-    final mockStream = StreamController<RealtimeMessage>.broadcast();
+    final mockStream = MockStream<RealtimeMessage>();
     final mockCloseFunction = () async {};
-    final subscription = RealtimeSubscription(
-        controller: mockStream,
-        close: mockCloseFunction,
-        channels: ['documents']);
+    final subscription = RealtimeSubscription(stream: mockStream, close: mockCloseFunction);
 
     test('should have the correct stream and close function', () {
-      expect(subscription.controller, equals(mockStream));
-      expect(subscription.stream, equals(mockStream.stream));
+      expect(subscription.stream, equals(mockStream));
       expect(subscription.close, equals(mockCloseFunction));
     });
   });