feat: use single isolate for functions and postgrest (#169)

* use single isolate for functions and postgrest

* initialize and dispose isolate

* dispose supabase client in tests

* flip the order within dispose

* await for stream cancel in tests

* force close mock server

* remove initializing

* Fix failing Github actions tests on isolate (#170)

* add initialize back in

* run tests against all pr

* add a small delay after each test case

* add option to receive isolate when creating client

* add a longer delay between tests

* remove unnecessary delay

* set web socket channel version

* Update lib/src/supabase_client.dart

Co-authored-by: Bruno D'Luka <[email protected]>

* add comments about isolate parameter

* Update lib/src/supabase_client.dart

Co-authored-by: Bruno D'Luka <[email protected]>

.github/workflows/main.yml

@@ -4,7 +4,6 @@ on:
   push:
     branches: [main]
   pull_request:
-    branches: [main]
 
 jobs:
   test:

lib/src/supabase_client.dart

@@ -8,6 +8,7 @@ import 'package:realtime_client/realtime_client.dart';
 import 'package:storage_client/storage_client.dart';
 import 'package:supabase/src/constants.dart';
 import 'package:supabase/src/supabase_query_builder.dart';
+import 'package:yet_another_json_isolate/yet_another_json_isolate.dart';
 
 class SupabaseClient {
   final String supabaseUrl;
@@ -26,6 +27,7 @@ class SupabaseClient {
   late final PostgrestClient rest;
   String? _changedAccessToken;
   late StreamSubscription<AuthState> _authStateSubscription;
+  late final YAJsonIsolate _isolate;
 
   /// Increment ID of the stream to create different realtime topic for each stream
   int _incrementId = 0;
@@ -45,6 +47,9 @@ class SupabaseClient {
   ///
   /// [storageRetryAttempts] specifies how many retry attempts there should be to
   ///  upload a file to Supabase storage when failed due to network interruption.
+  ///
+  /// Pass an instance of `YAJsonIsolate` to [isolate] to use your own persisted
+  /// isolate instance. A new instance will be created if [isolate] is omitted.
   SupabaseClient(
     this.supabaseUrl,
     this.supabaseKey, {
@@ -53,6 +58,7 @@ class SupabaseClient {
     Map<String, String> headers = Constants.defaultHeaders,
     Client? httpClient,
     int storageRetryAttempts = 0,
+    YAJsonIsolate? isolate,
   })  : restUrl = '$supabaseUrl/rest/v1',
         realtimeUrl = '$supabaseUrl/realtime/v1'.replaceAll('http', 'ws'),
         authUrl = '$supabaseUrl/auth/v1',
@@ -64,7 +70,8 @@ class SupabaseClient {
         schema = schema ?? 'public',
         _headers = headers,
         _httpClient = httpClient,
-        _storageRetryAttempts = storageRetryAttempts {
+        _storageRetryAttempts = storageRetryAttempts,
+        _isolate = isolate ?? (YAJsonIsolate()..initialize()) {
     auth = _initSupabaseAuthClient(
       autoRefreshToken: autoRefreshToken,
       headers: headers,
@@ -79,6 +86,7 @@ class SupabaseClient {
         functionsUrl,
         _getAuthHeaders(),
         httpClient: _httpClient,
+        isolate: _isolate,
       );
 
   /// Supabase Storage allows you to manage user-generated content, such as photos or videos.
@@ -101,6 +109,7 @@ class SupabaseClient {
       table: table,
       httpClient: _httpClient,
       incrementId: _incrementId,
+      isolate: _isolate,
     );
   }
 
@@ -137,8 +146,9 @@ class SupabaseClient {
     return realtime.removeAllChannels();
   }
 
-  void dispose() {
-    _authStateSubscription.cancel();
+  Future<void> dispose() async {
+    await _authStateSubscription.cancel();
+    await _isolate.dispose();
   }
 
   GoTrueClient _initSupabaseAuthClient({

lib/src/supabase_query_builder.dart

@@ -1,6 +1,7 @@
 import 'package:http/http.dart';
 import 'package:supabase/src/supabase_stream_builder.dart';
 import 'package:supabase/supabase.dart';
+import 'package:yet_another_json_isolate/yet_another_json_isolate.dart';
 
 class SupabaseQueryBuilder extends PostgrestQueryBuilder {
   final RealtimeClient _realtime;
@@ -16,6 +17,7 @@ class SupabaseQueryBuilder extends PostgrestQueryBuilder {
     required String table,
     Client? httpClient,
     required int incrementId,
+    required YAJsonIsolate isolate,
   })  : _realtime = realtime,
         _schema = schema,
         _table = table,
@@ -25,6 +27,7 @@ class SupabaseQueryBuilder extends PostgrestQueryBuilder {
           headers: headers,
           schema: schema,
           httpClient: httpClient,
+          isolate: isolate,
         );
 
   /// Notifies of data at the queried table

pubspec.yaml

@@ -8,15 +8,16 @@ environment:
   sdk: '>=2.15.0 <3.0.0'
 
 dependencies:
-  functions_client: ^1.0.0
+  functions_client: ^1.0.2
   gotrue: ^1.4.0
   http: ^0.13.4
   postgrest: ^1.2.1
   realtime_client: ^1.0.2
   storage_client: ^1.2.0
   rxdart: ^0.27.5
+  yet_another_json_isolate: ^1.0.1
 
 dev_dependencies:
   lints: ^1.0.1
   test: ^1.17.9
-  web_socket_channel: ^2.2.0
+  web_socket_channel: 2.2.0

test/client_test.dart

@@ -11,6 +11,10 @@ void main() {
       client = SupabaseClient(supabaseUrl, supabaseKey);
     });
 
+    tearDown(() async {
+      await client.dispose();
+    });
+
     test('X-Client-Info header is set properly on realtime', () {
       final xClientHeaderBeforeSlash =
           client.realtime.headers['X-Client-Info']!.split('/').first;

test/mock_test.dart

@@ -353,8 +353,10 @@ void main() {
   });
 
   tearDown(() async {
-    listener?.cancel();
     listeners.clear();
+    await listener?.cancel();
+
+    await client.dispose();
 
     // Wait for the realtime updates to come through
     await Future.delayed(Duration(milliseconds: 100));

test/realtime_test.dart

@@ -1,3 +1,4 @@
+import 'dart:async';
 import 'dart:io';
 
 import 'package:supabase/supabase.dart';
@@ -8,12 +9,14 @@ void main() {
   late HttpServer mockServer;
   late SupabaseClient client;
   late RealtimeChannel channel;
+  late StreamSubscription<WebSocket> subscription;
 
   group('Realtime subscriptions: ', () {
     setUp(() async {
       mockServer = await HttpServer.bind('localhost', 0);
 
-      mockServer.transform(WebSocketTransformer()).listen((webSocket) {
+      subscription =
+          mockServer.transform(WebSocketTransformer()).listen((webSocket) {
         final channel = IOWebSocketChannel(webSocket);
         channel.stream.listen((request) {
           channel.sink.add(request);
@@ -29,8 +32,10 @@ void main() {
     });
 
     tearDown(() async {
+      await client.dispose();
       await client.removeAllChannels();
-      await mockServer.close();
+      await subscription.cancel();
+      await mockServer.close(force: true);
     });
 
     /// subscribe on existing subscription fail