isolates - test observers work fine

currently broken: Cannot invoke callback on incorrect isolate.

Author: vaind

test/isolates_test.dart

@@ -4,9 +4,9 @@ import 'dart:isolate';
 
 import 'package:objectbox/src/bindings/bindings.dart';
 import 'package:test/test.dart';
+import 'package:objectbox/observable.dart';
 
 import 'entity.dart';
-import 'objectbox.g.dart';
 import 'test_env.dart';
 
 // We want to have types explicit - verifying the return types of functions.
@@ -15,8 +15,8 @@ void main() {
   /// Set up a simple echo isolate with request-response communication.
   /// This isn't really a test, just an example of how isolates can communicate.
   test('isolates two-way communication example', () async {
-    var receivePort = ReceivePort();
-    await Isolate.spawn(echoIsolate, receivePort.sendPort);
+    final receivePort = ReceivePort();
+    final isolate = await Isolate.spawn(echoIsolate, receivePort.sendPort);
 
     Completer sendPortCompleter = Completer<SendPort>();
     Completer responseCompleter;
@@ -41,12 +41,15 @@ void main() {
     // Send a message to the isolate
     expect(await call('hello'), equals('re:hello'));
     expect(await call('foo'), equals('re:foo'));
+
+    isolate.kill(priority: Isolate.immediate);
+    receivePort.close();
   });
 
   /// Work with a single store accross multiple isolates
   test('single store in multiple isolates', () async {
-    var receivePort = ReceivePort();
-    await Isolate.spawn(createDataIsolate, receivePort.sendPort);
+    final receivePort = ReceivePort();
+    final isolate = await Isolate.spawn(createDataIsolate, receivePort.sendPort);
 
     final sendPortCompleter = Completer<SendPort>();
     Completer<dynamic> responseCompleter;
@@ -72,22 +75,38 @@ void main() {
     final env = TestEnv('isolates');
     expect(await call(env.store.ptr.address), equals('store set'));
 
-    expect(env.box.isEmpty(), isTrue);
-    expect(await call(['put', 'Foo']), equals(1)); // returns inserted id = 1
-    expect(env.box.get(1).tString, equals('Foo'));
+    {
+      // check simple box operations
+      expect(env.box.isEmpty(), isTrue);
+      expect(await call(['put', 'Foo']), equals(1)); // returns inserted id = 1
+      expect(env.box.get(1).tString, equals('Foo'));
+    }
+
+    {
+      // verify that query streams (using observers) work fine across isolates
+      final queryStream = env.box.query().build().findStream();
+      expect(await call(['put', 'Bar']), equals(2));
+      List<TestEntity> found =
+          await queryStream.first.timeout(Duration(seconds: 1));
+      expect(found.length, equals(2));
+      expect(found.last.tString, equals('Bar'));
+    }
+
+    isolate.kill(priority: Isolate.immediate);
+    receivePort.close();
   });
 }
 
 // Echoes back any received message.
 void echoIsolate(SendPort sendPort) async {
   // Open the ReceivePort to listen for incoming messages
-  var port = ReceivePort();
+  final port = ReceivePort();
 
   // Send the port where the main isolate can contact us
   sendPort.send(port.sendPort);
 
   // Listen for messages
-  await for (var data in port) {
+  await for (final data in port) {
     // `data` is the message received.
     print('Isolate received: $data');
     sendPort.send('re:$data');
@@ -97,14 +116,14 @@ void echoIsolate(SendPort sendPort) async {
 // Creates data in the background, in the [Store] received as the first message.
 void createDataIsolate(SendPort sendPort) async {
   // Open the ReceivePort to listen for incoming messages
-  var port = ReceivePort();
+  final port = ReceivePort();
 
   // Send the port where the main isolate can contact us
   sendPort.send(port.sendPort);
 
   TestEnv env;
   // Listen for messages
-  await for (var data in port) {
+  await for (final data in port) {
     if (env == null) {
       // first message data is Store's C pointer address
       env = TestEnv.fromPtr(Pointer<OBX_store>.fromAddress(data));