Make streams emit errors when they can't fetch data (#233)

moor/CHANGELOG.md

@@ -9,6 +9,7 @@
 - New `batch` method on generated databases to execute multiple queries in a single batch
 - Experimental support to run moor on a background isolate
 - Reduce use of parentheses in SQL code generated at runtime
+- Query streams now emit errors that happened while running the query
 
 ## 2.0.1
 

moor/lib/src/runtime/executor/stream_queries.dart

@@ -200,11 +200,18 @@ class QueryStream<T> {
     // Fetch data if it's needed, publish that data if it's possible.
     if (!_controller.hasListener) return;
 
-    final data = await _fetcher.fetchData();
-    _lastData = data;
+    T data;
 
-    if (!_controller.isClosed) {
-      _controller.add(data);
+    try {
+      data = await _fetcher.fetchData();
+      _lastData = data;
+      if (!_controller.isClosed) {
+        _controller.add(data);
+      }
+    } catch (e, s) {
+      if (!_controller.isClosed) {
+        _controller.addError(e, s);
+      }
     }
   }
 }

moor/test/streams_test.dart

@@ -141,6 +141,15 @@ void main() {
     verify(executor.runSelect(any, any)).called(2);
   });
 
+  test('stream emits error when loading the query throws', () {
+    final exception = Exception('stub');
+    when(executor.runSelect(any, any))
+        .thenAnswer((_) => Future.error(exception));
+
+    final result = db.customSelectQuery('select 1').watch().first;
+    expectLater(result, throwsA(exception));
+  });
+
   group('stream keys', () {
     final keyA = StreamKey('SELECT * FROM users;', [], User);
     final keyB = StreamKey('SELECT * FROM users;', [], User);