fix: Consolidate realtime subscription for stream

supabase · Dec 10, 2024 · 668cfaa · 668cfaa
1 parent ccfcbf5
commit 668cfaa
Showing 1 changed file with 34 additions and 36 deletions.
diff --git a/packages/supabase/lib/src/supabase_stream_builder.dart b/packages/supabase/lib/src/supabase_stream_builder.dart
@@ -171,46 +171,44 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
 
     _channel!
         .onPostgresChanges(
-            event: PostgresChangeEvent.insert,
+            event: PostgresChangeEvent.all,
             schema: _schema,
             table: _table,
             filter: realtimeFilter,
             callback: (payload) {
-              final newRecord = payload.newRecord;
-              _streamData.add(newRecord);
-              _addStream();
-            })
-        .onPostgresChanges(
-            event: PostgresChangeEvent.update,
-            schema: _schema,
-            table: _table,
-            filter: realtimeFilter,
-            callback: (payload) {
-              final updatedIndex = _streamData.indexWhere(
-                (element) => _isTargetRecord(record: element, payload: payload),
-              );
-
-              final updatedRecord = payload.newRecord;
-              if (updatedIndex >= 0) {
-                _streamData[updatedIndex] = updatedRecord;
-              } else {
-                _streamData.add(updatedRecord);
-              }
-              _addStream();
-            })
-        .onPostgresChanges(
-            event: PostgresChangeEvent.delete,
-            schema: _schema,
-            table: _table,
-            filter: realtimeFilter,
-            callback: (payload) {
-              final deletedIndex = _streamData.indexWhere(
-                (element) => _isTargetRecord(record: element, payload: payload),
-              );
-              if (deletedIndex >= 0) {
-                /// Delete the data from in memory cache if it was found
-                _streamData.removeAt(deletedIndex);
-                _addStream();
+              switch (payload.eventType) {
+                case PostgresChangeEvent.insert:
+                  final newRecord = payload.newRecord;
+                  _streamData.add(newRecord);
+                  _addStream();
+                  break;
+                case PostgresChangeEvent.update:
+                  final updatedIndex = _streamData.indexWhere(
+                    (element) =>
+                        _isTargetRecord(record: element, payload: payload),
+                  );
+
+                  final updatedRecord = payload.newRecord;
+                  if (updatedIndex >= 0) {
+                    _streamData[updatedIndex] = updatedRecord;
+                  } else {
+                    _streamData.add(updatedRecord);
+                  }
+                  _addStream();
+                  break;
+                case PostgresChangeEvent.delete:
+                  final deletedIndex = _streamData.indexWhere(
+                    (element) =>
+                        _isTargetRecord(record: element, payload: payload),
+                  );
+                  if (deletedIndex >= 0) {
+                    /// Delete the data from in memory cache if it was found
+                    _streamData.removeAt(deletedIndex);
+                    _addStream();
+                  }
+                  break;
+                default:
+                  break;
               }
             })
         .subscribe((status, [error]) {