Isolate stream: replace existing stream implementation (#8)

Author: greenrobot-team

benchmark/bin/query.dart

@@ -8,7 +8,6 @@ void main() async {
   await QueryFind().report();
   await QueryFindIds().report();
   await QueryStream().report();
-  await QueryStreamIsolate().report();
 }
 
 class QueryBenchmark extends DbBenchmark {
@@ -57,21 +56,10 @@ class QueryFindIds extends QueryBenchmark {
   Future<void> run() async => query.findIds();
 }
 
-/// Stream where visitor is running in native code.
+/// Stream where visitor is running in Dart isolate.
 class QueryStream extends QueryBenchmark {
   QueryStream() : super('${QueryStream}');
 
   @override
   Future<void> run() async => await query.stream().toList();
 }
-
-/// Stream where visitor is running in Dart isolate.
-class QueryStreamIsolate extends QueryBenchmark {
-  QueryStreamIsolate() : super('${QueryStreamIsolate}');
-
-  @override
-  Future<void> run() async {
-    var stream = await query.streamAsync();
-    await stream.toList();
-  }
-}

objectbox/CHANGELOG.md

@@ -7,6 +7,7 @@
   Linux, Windows). This is where the [`install.sh`](/install.sh) script downloads it by default.
   E.g. it is no longer necessary to install the library globally to run `dart test` or `flutter test`.
 * Windows: Support database directory paths that contain unicode (UTF-8) characters. #406
+* Changed `Query.stream` to collect results in a worker isolate, which should typically be faster.
 * Update: [objectbox-c 0.16.0](https://github.com/objectbox/objectbox-c/releases/tag/v0.16.0).
 * Update: [objectbox-android 3.1.3](https://github.com/objectbox/objectbox-java/releases/tag/V3.1.3).
 * Add new [task with tag list Flutter example app](example/flutter/objectbox_demo_relations) that 

objectbox/lib/src/native/query/query.dart

@@ -818,69 +818,61 @@ class Query<T> {
     return result;
   }
 
-  /// Finds Objects matching the query, streaming them while the query executes.
-  ///
-  /// Note: make sure you evaluate performance in your use case - streams come
-  /// with an overhead so a plain [find()] is usually faster.
-  Stream<T> stream() => _stream1();
-
   /// Finds Objects matching the query, streaming them while the query executes.
   ///
   /// Results are streamed from a worker isolate in batches (the stream still
   /// returns objects one by one).
-  ///
-  /// This is typically faster than [stream()] and even [find()].
-  @experimental
-  Stream<T> streamAsync() => _streamIsolate();
+  Stream<T> stream() => _streamIsolate();
 
   /// Stream items by sending full flatbuffers binary as a message.
-  Stream<T> _stream1() {
-    initializeDartAPI();
-    final port = ReceivePort();
-    final cStream = checkObxPtr(
-        C.dartc_query_find(_cQuery, port.sendPort.nativePort), 'query stream');
-
-    var closed = false;
-    final close = () {
-      if (closed) return;
-      closed = true;
-      C.dartc_stream_close(cStream);
-      port.close();
-      reachabilityFence(this);
-    };
-
-    try {
-      final controller = StreamController<T>(onCancel: close);
-      port.listen((dynamic message) {
-        // We expect Uint8List for data and NULL when the query has finished.
-        if (message is Uint8List) {
-          try {
-            controller.add(
-                _entity.objectFromFB(_store, ByteData.view(message.buffer)));
-            return;
-          } catch (e) {
-            controller.addError(e);
-          }
-        } else if (message is String) {
-          controller.addError(
-              ObjectBoxException('Query stream native exception: $message'));
-        } else if (message != null) {
-          controller.addError(ObjectBoxException(
-              'Query stream received an invalid message type '
-              '(${message.runtimeType}): $message'));
-        }
-        // Close the stream, this will call the onCancel function.
-        // Do not call the onCancel function manually,
-        // if cancel() is called on the Stream subscription right afterwards it
-        // will use the shortcut in the onCancel function and not wait.
-        controller.close(); // done
-      });
-      return controller.stream;
-    } catch (e) {
-      close();
-      rethrow;
-    }
-  }
+  /// Replaced by _streamIsolate which in benchmarks has been faster.
+  // Stream<T> _stream1() {
+  //   initializeDartAPI();
+  //   final port = ReceivePort();
+  //   final cStream = checkObxPtr(
+  //       C.dartc_query_find(_cQuery, port.sendPort.nativePort), 'query stream');
+  //
+  //   var closed = false;
+  //   final close = () {
+  //     if (closed) return;
+  //     closed = true;
+  //     C.dartc_stream_close(cStream);
+  //     port.close();
+  //     reachabilityFence(this);
+  //   };
+  //
+  //   try {
+  //     final controller = StreamController<T>(onCancel: close);
+  //     port.listen((dynamic message) {
+  //       // We expect Uint8List for data and NULL when the query has finished.
+  //       if (message is Uint8List) {
+  //         try {
+  //           controller.add(
+  //               _entity.objectFromFB(_store, ByteData.view(message.buffer)));
+  //           return;
+  //         } catch (e) {
+  //           controller.addError(e);
+  //         }
+  //       } else if (message is String) {
+  //         controller.addError(
+  //             ObjectBoxException('Query stream native exception: $message'));
+  //       } else if (message != null) {
+  //         controller.addError(ObjectBoxException(
+  //             'Query stream received an invalid message type '
+  //             '(${message.runtimeType}): $message'));
+  //       }
+  //       // Close the stream, this will call the onCancel function.
+  //       // Do not call the onCancel function manually,
+  //       // if cancel() is called on the Stream subscription right afterwards it
+  //       // will use the shortcut in the onCancel function and not wait.
+  //       controller.close(); // done
+  //     });
+  //     return controller.stream;
+  //   } catch (e) {
+  //     close();
+  //     rethrow;
+  //   }
+  // }
 
   /// Stream items by sending pointers from native code.
   /// Interestingly this is slower even though it transfers only pointers...

objectbox/test/query_test.dart

@@ -665,7 +665,7 @@ void main() {
     q.close();
   });
 
-  testStream({required bool useIsolateStream}) async {
+  test('stream items', () async {
     final count = env.short ? 100 : 1000;
     final items = List<TestEntity>.generate(
         count, (i) => TestEntity.filled(id: 0, tByte: i % 30));
@@ -678,7 +678,7 @@ void main() {
     expect(query.count(), countMatching);
 
     final foundIds = query.findIds();
-    final stream = useIsolateStream ? query.streamAsync() : query.stream();
+    final stream = query.stream();
     final streamed = await stream.toList();
     expect(streamed.length, countMatching);
     final streamedIds = streamed.map((e) => e.id).toList(growable: false);
@@ -690,7 +690,7 @@ void main() {
     final streamListenedItems = <TestEntity>{};
 
     final start = DateTime.now();
-    final subStream = useIsolateStream ? query.streamAsync() : query.stream();
+    final subStream = query.stream();
     final subscription = subStream.listen(streamListenedItems.add);
     // Note: no upper limit, global test timeout will stop if it takes too long.
     int millis = 1;
@@ -703,14 +703,6 @@ void main() {
     expect(streamListenedItems.length, isNonZero);
 
     query.close();
-  }
-
-  test('stream items', () async {
-    await testStream(useIsolateStream: false);
-  });
-
-  test('stream items via isolate', () async {
-    await testStream(useIsolateStream: true);
   });
 
   test('set param single', () async {