fix(bloc_concurrency): close droppable stream when source ends mid-event

packages/bloc_concurrency/lib/src/droppable.dart

@@ -21,6 +21,7 @@ class _ExhaustMapStreamTransformer<T> extends StreamTransformerBase<T, T> {
   Stream<T> bind(Stream<T> stream) {
     late StreamSubscription<T> subscription;
     StreamSubscription<T>? mappedSubscription;
+    var isSourceDone = false;
 
     final controller = StreamController<T>(
       onCancel: () async {
@@ -30,6 +31,12 @@ class _ExhaustMapStreamTransformer<T> extends StreamTransformerBase<T, T> {
       sync: true,
     );
 
+    // Close the output stream once the source has completed and there is no
+    // event currently in flight.
+    void maybeClose() {
+      if (isSourceDone && mappedSubscription == null) controller.close();
+    }
+
     subscription = stream.listen(
       (data) {
         if (mappedSubscription != null) return;
@@ -39,11 +46,17 @@ class _ExhaustMapStreamTransformer<T> extends StreamTransformerBase<T, T> {
         mappedSubscription = mappedStream.listen(
           controller.add,
           onError: controller.addError,
-          onDone: () => mappedSubscription = null,
+          onDone: () {
+            mappedSubscription = null;
+            maybeClose();
+          },
         );
       },
       onError: controller.addError,
-      onDone: () => mappedSubscription ?? controller.close(),
+      onDone: () {
+        isSourceDone = true;
+        maybeClose();
+      },
     );
 
     return controller.stream;

packages/bloc_concurrency/test/src/droppable_test.dart

@@ -105,5 +105,39 @@ void main() {
 
       expect(states, isEmpty);
     });
+
+    test('closes the output stream when the source ends mid-event', () async {
+      final states = <int>[];
+      var isDone = false;
+      final controller = StreamController<int>();
+      final stream = droppable<int>()(controller.stream, (x) async* {
+        await wait();
+        yield x;
+      });
+
+      final subscription = stream.listen(
+        states.add,
+        onDone: () => isDone = true,
+      );
+
+      controller.add(0);
+
+      await tick();
+
+      // The event handler is in flight when the source stream completes.
+      await controller.close();
+
+      expect(isDone, isFalse);
+      expect(states, isEmpty);
+
+      await wait();
+      await tick();
+
+      // The in-flight event is still delivered and the output stream closes.
+      expect(states, equals([0]));
+      expect(isDone, isTrue);
+
+      await subscription.cancel();
+    });
   });
 }