fix: stream filter other than eq is not properly applied. (#156)

* fix: add tests to make sure rest api request from stream has correct filter

* fix: test properly makes request

* stream filters are properly applied

* proper filter is applied to realtime listener within stream

* fix: failing test

* can efficiently listen to multiple streams

* adds test to make sure listening to the same stream twice is supported

* update postgrest

* fix docs of stream

* use Behavior for the stream controller

* adds test on stream

Author: dshukertjr

lib/src/supabase_query_builder.dart

@@ -29,13 +29,13 @@ class SupabaseQueryBuilder extends PostgrestQueryBuilder {
   /// [primaryKey] list of name of primary key column(s).
   ///
   /// ```dart
-  /// supabase.from('chats').stream(['my_primary_key']).execute().listen(_onChatsReceived);
+  /// supabase.from('chats').stream(primaryKey: ['my_primary_key']).execute().listen(_onChatsReceived);
   /// ```
   ///
   /// `eq`, `order`, `limit` filter are available to limit the data being queried.
   ///
   /// ```dart
-  /// supabase.from('chats:room_id=eq.123').stream(['my_primary_key']).order('created_at').limit(20).execute().listen(_onChatsReceived);
+  /// supabase.from('chats:room_id=eq.123').stream(primaryKey: ['my_primary_key']).order('created_at').limit(20).execute().listen(_onChatsReceived);
   /// ```
   SupabaseStreamBuilder stream({required List<String> primaryKey}) {
     assert(primaryKey.isNotEmpty, 'Please specify primary key column(s).');

lib/src/supabase_stream_builder.dart

@@ -1,5 +1,6 @@
 import 'dart:async';
 
+import 'package:rxdart/rxdart.dart';
 import 'package:supabase/supabase.dart';
 
 enum _FilterType { eq, neq, lt, lte, gt, gte }
@@ -41,8 +42,6 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
 
   RealtimeChannel? _channel;
 
-  PostgrestBuilder? _postgrestBuilder;
-
   final String _schema;
 
   final String _table;
@@ -51,7 +50,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   final List<String> _uniqueColumns;
 
   /// StreamController for `stream()` method.
-  StreamController<SupabaseStreamEvent>? _streamController;
+  BehaviorSubject<SupabaseStreamEvent>? _streamController;
 
   /// Contains the combined data of postgrest and realtime to emit as stream.
   SupabaseStreamEvent _streamData = [];
@@ -84,7 +83,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   /// Only one filter can be applied to `.stream()`.
   ///
   /// ```dart
-  /// supabase.from('users').stream(['id']).eq('name', 'Supabase');
+  /// supabase.from('users').stream(primaryKey: ['id']).eq('name', 'Supabase');
   /// ```
   SupabaseStreamBuilder eq(String column, dynamic value) {
     assert(
@@ -104,7 +103,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   /// Only one filter can be applied to `.stream()`.
   ///
   /// ```dart
-  /// supabase.from('users').stream(['id']).neq('name', 'Supabase');
+  /// supabase.from('users').stream(primaryKey: ['id']).neq('name', 'Supabase');
   /// ```
   SupabaseStreamBuilder neq(String column, dynamic value) {
     assert(
@@ -124,7 +123,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   /// Only one filter can be applied to `.stream()`.
   ///
   /// ```dart
-  /// supabase.from('users').stream(['id']).lt('likes', 100);
+  /// supabase.from('users').stream(primaryKey: ['id']).lt('likes', 100);
   /// ```
   SupabaseStreamBuilder lt(String column, dynamic value) {
     assert(
@@ -144,7 +143,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   /// Only one filter can be applied to `.stream()`.
   ///
   /// ```dart
-  /// supabase.from('users').stream(['id']).lte('likes', 100);
+  /// supabase.from('users').stream(primaryKey: ['id']).lte('likes', 100);
   /// ```
   SupabaseStreamBuilder lte(String column, dynamic value) {
     assert(
@@ -164,7 +163,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   /// Only one filter can be applied to `.stream()`.
   ///
   /// ```dart
-  /// supabase.from('users').stream(['id']).gt('likes', '100');
+  /// supabase.from('users').stream(primaryKey: ['id']).gt('likes', '100');
   /// ```
   SupabaseStreamBuilder gt(String column, dynamic value) {
     assert(
@@ -184,7 +183,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   /// Only one filter can be applied to `.stream()`.
   ///
   /// ```dart
-  /// supabase.from('users').stream(['id']).gte('likes', 100);
+  /// supabase.from('users').stream(primaryKey: ['id']).gte('likes', 100);
   /// ```
   SupabaseStreamBuilder gte(String column, dynamic value) {
     assert(
@@ -204,7 +203,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   /// When `ascending` value is true, the result will be in ascending order.
   ///
   /// ```dart
-  /// supabase.from('users').stream(['id']).order('username', ascending: false);
+  /// supabase.from('users').stream(primaryKey: ['id']).order('username', ascending: false);
   /// ```
   SupabaseStreamBuilder order(String column, {bool ascending = false}) {
     _orderBy = _Order(column: column, ascending: ascending);
@@ -214,7 +213,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   /// Limits the result with the specified `count`.
   ///
   /// ```dart
-  /// supabase.from('users').stream(['id']).limit(10);
+  /// supabase.from('users').stream(primaryKey: ['id']).limit(10);
   /// ```
   SupabaseStreamBuilder limit(int count) {
     _limit = count;
@@ -223,13 +222,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
 
   @Deprecated('Directly listen without execute instead. Deprecated in 1.0.0')
   Stream<SupabaseStreamEvent> execute() {
-    _streamController = StreamController.broadcast(
-      onCancel: () {
-        _channel?.unsubscribe();
-        _streamController?.close();
-      },
-    );
-    _getStreamData();
+    _setupStream();
     return _streamController!.stream;
   }
 
@@ -240,37 +233,27 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
     void Function()? onDone,
     bool? cancelOnError,
   }) {
-    if (_postgrestBuilder == null) {
-      PostgrestFilterBuilder query = _queryBuilder.select();
-      if (_streamFilter != null) {
-        query = query.eq(_streamFilter!.column, _streamFilter!.value);
-      }
-      PostgrestTransformBuilder? transformQuery;
-      if (_orderBy != null) {
-        transformQuery =
-            query.order(_orderBy!.column, ascending: _orderBy!.ascending);
-      }
-      if (_limit != null) {
-        transformQuery = (transformQuery ?? query).limit(_limit!);
-      }
-      _postgrestBuilder = transformQuery ?? query;
-    }
+    _setupStream();
+    return _streamController!.stream.listen(
+      onData,
+      onError: onError,
+      onDone: onDone,
+      cancelOnError: cancelOnError,
+    );
+  }
 
-    _streamController = StreamController.broadcast(
+  /// Sets up the stream controller and calls the method to get data as necessary
+  void _setupStream() {
+    _streamController ??= BehaviorSubject(
       onListen: () {
         _getStreamData();
       },
       onCancel: () {
         _channel?.unsubscribe();
         _streamController?.close();
+        _streamController = null;
       },
     );
-    return _streamController!.stream.listen(
-      onData,
-      onError: onError,
-      onDone: onDone,
-      cancelOnError: cancelOnError,
-    );
   }
 
   Future<void> _getStreamData() async {
@@ -300,9 +283,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
           event: 'UPDATE',
           schema: _schema,
           table: _table,
-          filter: _streamFilter != null
-              ? '${_streamFilter!.column}=eq.${_streamFilter!.value}'
-              : null,
+          filter: realtimeFilter,
         ), (payload, [ref]) {
       final updatedIndex = _streamData.indexWhere(
         (element) => _isTargetRecord(record: element, payload: payload),
@@ -320,9 +301,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
           event: 'DELETE',
           schema: _schema,
           table: _table,
-          filter: _streamFilter != null
-              ? '${_streamFilter!.column}=eq.${_streamFilter!.value}'
-              : null,
+          filter: realtimeFilter,
         ), (payload, [ref]) {
       final deletedIndex = _streamData.indexWhere(
         (element) => _isTargetRecord(record: element, payload: payload),
@@ -368,7 +347,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
     }
 
     try {
-      final data = await _postgrestBuilder;
+      final data = await (transformQuery ?? query);
       final rows = SupabaseStreamEvent.from(data as List);
       _streamData.addAll(rows);
       _addStream();

pubspec.yaml

@@ -11,9 +11,10 @@ dependencies:
   functions_client: ^1.0.0
   gotrue: ^1.0.0
   http: ^0.13.4
-  postgrest: ^1.0.0
+  postgrest: ^1.0.1
   realtime_client: ^1.0.0
   storage_client: ^1.0.0
+  rxdart: ^0.27.5
 
 dev_dependencies:
   lints: ^1.0.1

test/mock_test.dart

@@ -21,7 +21,7 @@ void main() {
   /// `testFilter` is used to test incoming realtime filter. The value should match the realtime filter set by the library.
   Future<void> handleRequests(
     HttpServer server, {
-    String? testFilter,
+    String? expectedFilter,
   }) async {
     await for (final HttpRequest request in server) {
       final headers = request.headers;
@@ -35,6 +35,11 @@ void main() {
         if (foundApiKey == customApiKey) {
           expect(headers.value('customfield'), 'customvalue');
         }
+
+        // Check that rest api contains the correct filter in the URL
+        if (expectedFilter != null) {
+          expect(url.contains(expectedFilter), isTrue);
+        }
       }
       if (url == '/rest/v1/todos?select=task%2Cstatus') {
         final jsonString = jsonEncode([
@@ -87,6 +92,13 @@ void main() {
           ..headers.contentType = ContentType.json
           ..write(jsonString)
           ..close();
+      } else if (url.contains('rest')) {
+        // Just return an empty string as dummy data if any other rest request
+        request.response
+          ..statusCode = HttpStatus.ok
+          ..headers.contentType = ContentType.json
+          ..write('[]')
+          ..close();
       } else if (url.contains('realtime')) {
         webSocket = await WebSocketTransformer.upgrade(request);
         if (hasListener) {
@@ -107,8 +119,8 @@ void main() {
                   ['postgres_changes']
               .first['filter'];
 
-          if (testFilter != null) {
-            expect(realtimeFilter, testFilter);
+          if (expectedFilter != null) {
+            expect(realtimeFilter, expectedFilter);
           }
 
           final replyString = jsonEncode({
@@ -327,6 +339,23 @@ void main() {
 
         stream.listen(expectAsync1((event) {}, count: 4));
       });
+      test("can listen twice at the same time", () async {
+        final stream = client.from('todos').stream(primaryKey: ['id']);
+        stream.listen(expectAsync1((event) {}, count: 4));
+        stream.listen(expectAsync1((event) {}, count: 4));
+
+        // All realtime events are done emitting, so should receive the currnet data
+      });
+      test("stream should emit the last emitted data when listened to",
+          () async {
+        final stream = client.from('todos').stream(primaryKey: ['id']);
+        stream.listen(expectAsync1((event) {}, count: 4));
+
+        await Future.delayed(Duration(seconds: 3));
+
+        // All realtime events are done emitting, so should receive the currnet data
+        stream.listen(expectAsync1((event) {}, count: 1));
+      });
       test('emits data', () {
         final stream = client.from('todos').stream(primaryKey: ['id']);
         expect(
@@ -469,7 +498,7 @@ void main() {
 
   group('realtime filter', () {
     test('can filter stream results with eq', () {
-      handleRequests(mockServer, testFilter: 'status=eq.true');
+      handleRequests(mockServer, expectedFilter: 'status=eq.true');
       final stream =
           client.from('todos').stream(primaryKey: ['id']).eq('status', true);
       expect(
@@ -486,24 +515,39 @@ void main() {
       );
     });
 
+    test('can filter stream results with neq', () {
+      handleRequests(mockServer, expectedFilter: 'id=neq.2');
+      final stream =
+          client.from('todos').stream(primaryKey: ['id']).neq('id', 2);
+      expect(stream, emits(isList));
+    });
+
     test('can filter stream results with gt', () {
-      handleRequests(mockServer, testFilter: 'id=gt.2');
-      client.from('todos').stream(primaryKey: ['id']).gt('id', 2);
+      handleRequests(mockServer, expectedFilter: 'id=gt.2');
+      final stream =
+          client.from('todos').stream(primaryKey: ['id']).gt('id', 2);
+      expect(stream, emits(isList));
     });
 
     test('can filter stream results with gte', () {
-      handleRequests(mockServer, testFilter: 'id=gte.2');
-      client.from('todos').stream(primaryKey: ['id']).gte('id', 2);
+      handleRequests(mockServer, expectedFilter: 'id=gte.2');
+      final stream =
+          client.from('todos').stream(primaryKey: ['id']).gte('id', 2);
+      expect(stream, emits(isList));
     });
 
     test('can filter stream results with lt', () {
-      handleRequests(mockServer, testFilter: 'id=lt.2');
-      client.from('todos').stream(primaryKey: ['id']).lt('id', 2);
+      handleRequests(mockServer, expectedFilter: 'id=lt.2');
+      final stream =
+          client.from('todos').stream(primaryKey: ['id']).lt('id', 2);
+      expect(stream, emits(isList));
     });
 
     test('can filter stream results with lte', () {
-      handleRequests(mockServer, testFilter: 'id=lte.2');
-      client.from('todos').stream(primaryKey: ['id']).lte('id', 2);
+      handleRequests(mockServer, expectedFilter: 'id=lte.2');
+      final stream =
+          client.from('todos').stream(primaryKey: ['id']).lte('id', 2);
+      expect(stream, emits(isList));
     });
   });
 }