Fix closing after delay

Author: simolus3

packages/sqlite_async/lib/src/update_notification.dart

@@ -88,7 +88,13 @@ Stream<T> _throttleStream<T extends Object>(Stream<T> input, Duration timeout,
     if (!nextPing.isCompleted) {
       nextPing.complete();
     }
-  }, onDone: () => done = true);
+  }, onDone: () {
+    if (!nextPing.isCompleted) {
+      nextPing.complete();
+    }
+
+    done = true;
+  });
 
   try {
     if (addOne != null) {
@@ -101,6 +107,8 @@ Stream<T> _throttleStream<T extends Object>(Stream<T> input, Duration timeout,
       // If a value is available now, we'll use it immediately.
       // If not, this waits for it.
       await nextPing.future;
+      if (done) break;
+
       // Capture any new values coming in while we wait.
       nextPing = Completer<void>();
       T data = lastData as T;

packages/sqlite_async/test/update_notification_test.dart

@@ -112,6 +112,26 @@ void main() {
           expect(control.pendingTimers, isEmpty);
         });
       });
+
+      test('closes when source closes after delay', () {
+        fakeAsync((control) {
+          final source = StreamController<UpdateNotification>(sync: true)
+            ..onCancel = () => Future.value();
+          final events = <UpdateNotification>[];
+          var done = false;
+
+          UpdateNotification.throttleStream(source.stream, timeout)
+              .listen(events.add, onDone: () => done = true);
+
+          control.elapse(const Duration(hours: 1));
+          source.close();
+
+          control.flushTimers();
+          expect(events, isEmpty);
+          expect(done, isTrue);
+          expect(control.pendingTimers, isEmpty);
+        });
+      });
     });
 
     test('filter tables', () async {