From 23585ad9209d610cf7f4ac36e54007ce990ee424 Mon Sep 17 00:00:00 2001
From: Simon Binder <oss@simonbinder.eu>
Date: Wed, 19 Feb 2020 20:01:24 +0100
Subject: [PATCH] Don't await StreamController.close() when closing streams

---
 moor/CHANGELOG.md                                 |  2 ++
 moor/lib/src/runtime/executor/stream_queries.dart | 14 +++++++++++---
 moor/lib/src/runtime/executor/transactions.dart   |  4 +++-
 moor/test/streams_test.dart                       | 13 +++++++++++++
 4 files changed, 29 insertions(+), 4 deletions(-)

diff --git a/moor/CHANGELOG.md b/moor/CHANGELOG.md
index fbe0b8b4..7861f75c 100644
--- a/moor/CHANGELOG.md
+++ b/moor/CHANGELOG.md
@@ -15,6 +15,8 @@
 - __Breaking__: Remove `customSelectStream` from `QueryEngine`. The `customSelect`
   method now returns an `Selectable` (like `customSelectQuery`, which in turn has been deprecated).
 - Experimentally support IndexedDB to store sqlite data on the web
+- Moor will no longer wait for query stream listeners to receive a done event when closing a database
+  or transaction.
 
 ## 2.4.0
 
diff --git a/moor/lib/src/runtime/executor/stream_queries.dart b/moor/lib/src/runtime/executor/stream_queries.dart
index 206e51c5..15b6a9ec 100644
--- a/moor/lib/src/runtime/executor/stream_queries.dart
+++ b/moor/lib/src/runtime/executor/stream_queries.dart
@@ -5,6 +5,7 @@ import 'package:collection/collection.dart';
 import 'package:meta/meta.dart';
 import 'package:moor/moor.dart';
 import 'package:moor/src/utils/start_with_value_transformer.dart';
+import 'package:pedantic/pedantic.dart';
 
 const _listEquality = ListEquality<dynamic>();
 
@@ -161,8 +162,15 @@ class StreamQueryStore {
     _isShuttingDown = true;
 
     for (final stream in _activeKeyStreams.values) {
-      await stream._controller.close();
+      // Note: StreamController.close waits until the done event has been
+      // received by a subscriber. If there is a paused StreamSubscription on
+      // a query stream, this would pause forever. In particular, this is
+      // causing deadlocks in tests.
+      // https://github.com/dart-lang/test/issues/1183#issuecomment-588357154
+      unawaited(stream._controller.close());
     }
+    // awaiting this is fine - the stream is never exposed to users and we don't
+    // pause any subscriptions on it.
     await _updatedTableNames.close();
 
     while (_pendingTimers.isNotEmpty) {
@@ -251,7 +259,7 @@ class QueryStream<T> {
     }
   }
 
-  Future<void> close() {
-    return _controller.close();
+  void close() {
+    _controller.close();
   }
 }
diff --git a/moor/lib/src/runtime/executor/transactions.dart b/moor/lib/src/runtime/executor/transactions.dart
index 02a81962..375807c3 100644
--- a/moor/lib/src/runtime/executor/transactions.dart
+++ b/moor/lib/src/runtime/executor/transactions.dart
@@ -72,7 +72,9 @@ class _TransactionStreamStore extends StreamQueryStore {
     parent.handleTableUpdatesByName(affectedTables);
 
     await super.close();
-    await Future.wait(_queriesWithoutKey.map((e) => e.close()));
+    for (final query in _queriesWithoutKey) {
+      query.close();
+    }
   }
 }
 
diff --git a/moor/test/streams_test.dart b/moor/test/streams_test.dart
index 9f68394c..416f8ba5 100644
--- a/moor/test/streams_test.dart
+++ b/moor/test/streams_test.dart
@@ -162,6 +162,19 @@ void main() {
     expectLater(result, throwsA(exception));
   });
 
+  test('database can be closed when a stream has a paused subscription',
+      () async {
+    // this test is more relevant than it seems - some test stream matchers
+    // leave the stream in an empty state.
+    final stream = db.select(db.users).watch();
+    final subscription = stream.listen((_) {})..pause();
+
+    await db.close();
+
+    subscription.resume();
+    await subscription.cancel();
+  });
+
   group('stream keys', () {
     final keyA = StreamKey('SELECT * FROM users;', [], User);
     final keyB = StreamKey('SELECT * FROM users;', [], User);