diff --git a/packages/bloc_concurrency/lib/src/droppable.dart b/packages/bloc_concurrency/lib/src/droppable.dart index 56d80237434..a5522db1e22 100644 --- a/packages/bloc_concurrency/lib/src/droppable.dart +++ b/packages/bloc_concurrency/lib/src/droppable.dart @@ -21,6 +21,7 @@ class _ExhaustMapStreamTransformer extends StreamTransformerBase { Stream bind(Stream stream) { late StreamSubscription subscription; StreamSubscription? mappedSubscription; + var isSourceDone = false; final controller = StreamController( onCancel: () async { @@ -30,6 +31,12 @@ class _ExhaustMapStreamTransformer extends StreamTransformerBase { 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 extends StreamTransformerBase { 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; diff --git a/packages/bloc_concurrency/test/src/droppable_test.dart b/packages/bloc_concurrency/test/src/droppable_test.dart index 45d1e666cd6..80fae55e39c 100644 --- a/packages/bloc_concurrency/test/src/droppable_test.dart +++ b/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 = []; + var isDone = false; + final controller = StreamController(); + final stream = droppable()(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(); + }); }); }