diff --git a/CHANGELOG.md b/CHANGELOG.md index bacba0fef57..4bdeeb175c2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,10 @@ ### Libraries +#### `dart:async` + +- Added `broadcast` parameter to `Stream.empty` constructor. + #### `dart:developer` - Deprecated the `Service.getIsolateID` method. diff --git a/pkg/async_helper/lib/async_helper.dart b/pkg/async_helper/lib/async_helper.dart index 461e36a124a..bddbe38b20f 100644 --- a/pkg/async_helper/lib/async_helper.dart +++ b/pkg/async_helper/lib/async_helper.dart @@ -4,8 +4,8 @@ /// This library is used for testing asynchronous tests. /// If a test is asynchronous, it needs to notify the testing driver -/// about this (otherwise tests may get reported as passing [after main() -/// finished] even if the asynchronous operations fail). +/// about this (otherwise tests may get reported as passing after main() +/// finished even if the asynchronous operations fail). /// /// This library provides four methods /// - asyncStart(): Needs to be called before an asynchronous operation is @@ -16,10 +16,9 @@ /// - asyncTest(f()): Helper method that wraps a computation that returns a /// Future with matching calls to asyncStart() and /// asyncSuccess(_). -/// After the last asyncStart() called was matched with a corresponding +/// After every asyncStart() called is matched with a corresponding /// asyncEnd() or asyncSuccess(_) call, the testing driver will be notified that /// the tests is done. - library async_helper; import 'dart:async'; @@ -68,21 +67,25 @@ void asyncEnd() { /// Call this after an asynchronous test has ended successfully. This is a helper /// for calling [asyncEnd]. /// -/// This method intentionally has a signature that matches [:Future.then:] as a -/// convenience for calling [asyncEnd] when a [:Future:] completes without error, +/// This method intentionally has a signature that matches `Future.then` as a +/// convenience for calling [asyncEnd] when a `Future` completes without error, /// like this: -/// -/// asyncStart(); -/// Future result = test(); -/// result.then(asyncSuccess); -void asyncSuccess(_) => asyncEnd(); +/// ```dart +/// asyncStart(); +/// Future result = test(); +/// result.then(asyncSuccess); +/// ``` +void asyncSuccess(void _) { + asyncEnd(); +} -/// Helper method for performing asynchronous tests involving [:Future:]. +/// Helper method for performing asynchronous tests involving `Future`. /// -/// [f] must return a [:Future:] for the test computation. -Future asyncTest(Function() f) { +/// The function [test] must return a `Future` which completes without error +/// when the test is successful. +Future asyncTest(Function() test) { asyncStart(); - return f().then(asyncSuccess); + return test().then(asyncSuccess); } /// Verifies that the asynchronous [result] throws a [T]. @@ -109,7 +112,7 @@ Future asyncExpectThrows(Future result, // Handle null being passed in from legacy code // while also avoiding producing an unnecessary null check warning here. if ((reason as dynamic) == null) reason = ""; - // Only include the type in he message if it's not a top-type. + // Only include the type in the message if it's not `Object`. var type = Object() is! T ? "<$T>" : ""; return "asyncExpectThrows$type($reason):"; } diff --git a/sdk/lib/async/stream.dart b/sdk/lib/async/stream.dart index a9a925a0578..1b848e24fee 100644 --- a/sdk/lib/async/stream.dart +++ b/sdk/lib/async/stream.dart @@ -154,7 +154,15 @@ abstract mixin class Stream { /// }, /// ); /// ``` - const factory Stream.empty() = _EmptyStream; + /// + /// The stream defaults to being a broadcast stream, + /// as reported by [isBroadcast]. + /// This value can be changed by passing `false` as + /// the [broadcast] parameter, which defaults to `true`. + /// + /// The stream can be listened to more than once, + /// whether it reports itself as broadcast or not. + const factory Stream.empty({@Since("3.2") bool broadcast}) = _EmptyStream; /// Creates a stream which emits a single data event before closing. /// diff --git a/sdk/lib/async/stream_impl.dart b/sdk/lib/async/stream_impl.dart index 015d453c101..f32bc6fc605 100644 --- a/sdk/lib/async/stream_impl.dart +++ b/sdk/lib/async/stream_impl.dart @@ -630,49 +630,97 @@ typedef void _BroadcastCallback(StreamSubscription subscription); /// Done subscription that will send one done event as soon as possible. class _DoneStreamSubscription implements StreamSubscription { - static const int _DONE_SENT = 1; - static const int _SCHEDULED = 2; - static const int _PAUSED = 4; + // States of the subscription. + // + // The subscription will try to send a done event. + // When created, it schedules a microtask to emit the + // event to the current [_onDone] callback. + // If paused when the microtask happens, the subscription won't + // send the done event. It will reschedule a new microtask when + // the subscription is resumed. + // If cancelled, the event will not be sent, and a new + // microtask won't be scheduled, and further pauses + // are ignored. + /// Sending a done event is allowed. + /// + /// Not paused or cancelled, no microtask scheduled. + static const int _stateReadyToSend = 0; + + /// Set when a microtask is scheduled, and not cancelled. + static const int _stateScheduled = 1; + + /// State set when done event sent or subscription cancelled. + /// + /// Any negative value counts as done, so we can safely subtract + /// [_stateScheduled] or [_statePausedOnce] without affecting + /// being cancelled. + /// + /// Never considered paused in this state, may still be scheduled, + /// and [_onDone] is always set to `null`. + static const int _stateDone = -1; + + /// Added for each pause while not done or cancelled, subtracted on resume. + /// + /// Subscription is paused when state is at least `_statePausedOnce`. + static const int _statePausedOnce = 2; + + int _state; final Zone _zone; - int _state = 0; void Function()? _onDone; - _DoneStreamSubscription(this._onDone) : _zone = Zone.current { - _schedule(); - } - - bool get _isSent => (_state & _DONE_SENT) != 0; - bool get _isScheduled => (_state & _SCHEDULED) != 0; - bool get isPaused => _state >= _PAUSED; - - void _schedule() { - if (_isScheduled) return; - _zone.scheduleMicrotask(_sendDone); - _state |= _SCHEDULED; - } - - void onData(void handleData(T data)?) {} - void onError(Function? handleError) {} - void onDone(void handleDone()?) { - _onDone = handleDone; - } - - void pause([Future? resumeSignal]) { - _state += _PAUSED; - if (resumeSignal != null) resumeSignal.whenComplete(resume); - } - - void resume() { - if (isPaused) { - _state -= _PAUSED; - if (!isPaused && !_isSent) { - _schedule(); - } + _DoneStreamSubscription(void Function()? onDone) + : _zone = Zone.current, + _state = _stateScheduled { + scheduleMicrotask(_onMicrotask); + if (onDone != null) { + _onDone = _zone.registerCallback(onDone); } } - Future cancel() => Future._nullFuture; + bool get isPaused => _state >= _statePausedOnce; + + /// True after being cancelled, or after delivering done event. + static bool _isDone(int state) => state < 0; + + static int _incrementPauseCount(int state) => state + _statePausedOnce; + static int _decrementPauseCount(int state) => state - _statePausedOnce; + + void onData(void handleData(T data)?) {} + + void onError(Function? handleError) {} + + void onDone(void handleDone()?) { + if (!_isDone(_state)) { + if (handleDone != null) handleDone = _zone.registerCallback(handleDone); + _onDone = handleDone; + } + } + + void pause([Future? resumeSignal]) { + if (!_isDone(_state)) { + _state = _incrementPauseCount(_state); + if (resumeSignal != null) resumeSignal.whenComplete(resume); + } + } + + void resume() { + var resumeState = _decrementPauseCount(_state); + if (resumeState < 0) return; // Wasn't paused. + if (resumeState == _stateReadyToSend) { + // No longer paused, and not already scheduled. + _state = _stateScheduled; + scheduleMicrotask(_onMicrotask); + } else { + _state = resumeState; + } + } + + Future cancel() { + _state = _stateDone; + _onDone = null; + return Future._nullFuture; + } Future asFuture([E? futureValue]) { E resultValue; @@ -684,19 +732,28 @@ class _DoneStreamSubscription implements StreamSubscription { } else { resultValue = futureValue; } - _Future result = new _Future(); - _onDone = () { - result._completeWithValue(resultValue); - }; + _Future result = _Future(); + if (!_isDone(_state)) { + _onDone = _zone.registerCallback(() { + result._completeWithValue(resultValue); + }); + } return result; } - void _sendDone() { - _state &= ~_SCHEDULED; - if (isPaused) return; - _state |= _DONE_SENT; - var doneHandler = _onDone; - if (doneHandler != null) _zone.runGuarded(doneHandler); + void _onMicrotask() { + var unscheduledState = _state - _stateScheduled; + if (unscheduledState == _stateReadyToSend) { + // Send the done event. + _state = _stateDone; + if (_onDone case var doneHandler?) { + _onDone = null; + _zone.runGuarded(doneHandler); + } + } else { + // Paused or cancelled. + _state = unscheduledState; + } } } @@ -1007,11 +1064,11 @@ class _StreamIterator implements StreamIterator { /// An empty broadcast stream, sending a done event as soon as possible. class _EmptyStream extends Stream { - const _EmptyStream(); - bool get isBroadcast => true; - StreamSubscription listen(void onData(T data)?, - {Function? onError, void onDone()?, bool? cancelOnError}) { - return new _DoneStreamSubscription(onDone); + const _EmptyStream({bool broadcast = true}) : isBroadcast = broadcast; + final bool isBroadcast; + StreamSubscription listen(void Function(T data)? onData, + {Function? onError, void Function()? onDone, bool? cancelOnError}) { + return _DoneStreamSubscription(onDone); } } diff --git a/tests/lib/async/stream_empty_test.dart b/tests/lib/async/stream_empty_test.dart index d421b3de6fe..a5c6fdd3445 100644 --- a/tests/lib/async/stream_empty_test.dart +++ b/tests/lib/async/stream_empty_test.dart @@ -7,62 +7,158 @@ import "package:expect/expect.dart"; import "dart:async"; import 'package:async_helper/async_helper.dart'; -main() { - asyncStart(); - runTest().whenComplete(asyncEnd); +main() async { + await asyncTest(() async { + // Is a `const` constructor. + // Can be called with optional boolean to say whether broadcast or not. + await asyncTest(() => emptyTest(const Stream.empty(), true)); + await asyncTest( + () => emptyTest(const Stream.empty(broadcast: true), true)); + await asyncTest( + () => emptyTest(const Stream.empty(broadcast: false), false)); + + // Check that the behavior is consistent with other empty multi-subscription + // streams. + await asyncTest(() => emptyTest(Stream.fromIterable([]), false)); + await asyncTest(() => + emptyTest((StreamController.broadcast()..close()).stream, true)); + await asyncTest( + () => emptyTest(Stream.multi((c) => c.close()), false)); + }); + await flushMicrotasks(); } -Future runTest() async { - unreachable([a, b]) { - throw "UNREACHABLE"; +// Function which fails the test if it gets called. +void unreachable([a, b]) { + Expect.fail("Unreachable"); +} + +/// Check that something happens once. +class Checker { + bool checked = false; + void check() { + if (checked) Expect.fail("Checked more than once"); + checked = true; } +} - int tick = 0; - ticker() { - tick++; - } +Future emptyTest(Stream s, bool broadcast) async { + var checker = Checker(); + // Respects type parameter (not a `Stream`) + Expect.notType>(s); - asyncStart(); + // Has the expected `isBroadcast`. + Expect.equals(broadcast, s.isBroadcast); - Stream s = const Stream.empty(); // Is const constructor. - Expect.isFalse(s is Stream); // Respects type parameter. StreamSubscription sub = - s.listen(unreachable, onError: unreachable, onDone: ticker); - Expect.isFalse(sub is StreamSubscription); // Type parameter in sub. + s.listen(unreachable, onError: unreachable, onDone: checker.check); + // Type parameter of subscription repspects stream. + // Not a `StreamSubscription`. + Expect.isFalse(sub is StreamSubscription); - // Doesn't do callback in response to listen. - Expect.equals(tick, 0); + Expect.isFalse(sub.isPaused); + // Doesn't do callback immediately in response to listen. + Expect.isFalse(checker.checked); await flushMicrotasks(); - // Completes eventually. - Expect.equals(tick, 1); + // Completes eventually, after a microtask. + Expect.isTrue(checker.checked); + Expect.isFalse(sub.isPaused); - // It's a broadcast stream - can listen twice. - Expect.isTrue(s.isBroadcast); + // Can listen more than once, whether broadcast stream or not. + checker = Checker(); StreamSubscription sub2 = - s.listen(unreachable, onError: unreachable, onDone: unreachable); - // respects pause. + s.listen(unreachable, onError: unreachable, onDone: checker.check); + // Respects pauses. sub2.pause(); - await flushMicrotasks(); - // respects cancel. - sub2.cancel(); - await flushMicrotasks(); - Expect.equals(tick, 1); - // Still not complete. + Expect.isTrue(sub2.isPaused); + sub2.pause(); + Expect.isTrue(sub2.isPaused); + sub2.pause(Future.value(0)); + Expect.isTrue(sub2.isPaused); - StreamSubscription sub3 = - s.listen(unreachable, onError: unreachable, onDone: ticker); - // respects pause. + await flushMicrotasks(); + Expect.isTrue(sub2.isPaused); + Expect.isFalse(checker.checked); + + // Resumes when all pauses resumed. + sub2.resume(); + Expect.isTrue(sub2.isPaused); + await flushMicrotasks(); + Expect.isFalse(checker.checked); + Expect.isTrue(sub2.isPaused); + + sub2.resume(); + Expect.isFalse(sub2.isPaused); + await flushMicrotasks(); + Expect.isTrue(checker.checked); + Expect.isFalse(sub2.isPaused); + + // Can't pause after done. + sub2.pause(Future.value(0)); + Expect.isFalse(sub2.isPaused); + sub2.pause(); + Expect.isFalse(sub2.isPaused); + + // Respects cancel. + var sub3 = s.listen(unreachable, onError: unreachable, onDone: unreachable); + sub3.cancel(); + sub3.onDone(unreachable); + Expect.isFalse(sub3.isPaused); + await flushMicrotasks(); + Expect.isFalse(sub3.isPaused); + + // Can't pause after cancel sub3.pause(); - Expect.equals(tick, 1); - await flushMicrotasks(); - // Doesn't complete while paused. - Expect.equals(tick, 1); - sub3.resume(); - await flushMicrotasks(); - // Now completed. - Expect.equals(tick, 2); + Expect.isFalse(sub3.isPaused); + sub3.pause(Future.value(0)); + Expect.isFalse(sub3.isPaused); + // No errors. - asyncEnd(); + // Respects cancel while paused. + var sub4 = s.listen(unreachable, onError: unreachable, onDone: unreachable); + sub4.pause(); + sub4.cancel(); + sub4.onDone(unreachable); + await flushMicrotasks(); + sub4.resume(); + await flushMicrotasks(); + + // Check that the stream is zone-aware. + // Registers onDone callback. + var log = []; + late final Zone zone; + + void callback1() { + // Run in correct zone. + Expect.equals(zone, Zone.current); + log.add("don1"); + } + + void callback2() { + // Run in correct zone. + Expect.equals(zone, Zone.current); + log.add("don2"); + } + + zone = Zone.current.fork( + specification: ZoneSpecification(registerCallback: (s, p, z, f) { + if (f == callback1) log.add("reg1"); + if (f == callback2) log.add("reg2"); + return p.registerCallback(z, f); + }, run: (s, p, z, f) { + if (f == callback1) log.add("run1"); + if (f == callback2) log.add("run2"); + return p.run(z, f); + })); + + await zone.run(() async { + var s = Stream.empty(); + + var sub = s.listen(unreachable, onError: unreachable, onDone: callback1); + sub.onDone(callback2); + }); + await flushMicrotasks(); + Expect.listEquals(["reg1", "reg2", "run2", "don2"], log); } Future flushMicrotasks() => new Future.delayed(Duration.zero); diff --git a/tests/lib_2/async/stream_empty_test.dart b/tests/lib_2/async/stream_empty_test.dart index e4914a907a5..8390d0f2b03 100644 --- a/tests/lib_2/async/stream_empty_test.dart +++ b/tests/lib_2/async/stream_empty_test.dart @@ -9,62 +9,158 @@ import "package:expect/expect.dart"; import "dart:async"; import 'package:async_helper/async_helper.dart'; -main() { - asyncStart(); - runTest().whenComplete(asyncEnd); +main() async { + await asyncTest(() async { + // Is a `const` constructor. + // Can be called with optional boolean to say whether broadcast or not. + await asyncTest(() => emptyTest(const Stream.empty(), true)); + await asyncTest( + () => emptyTest(const Stream.empty(broadcast: true), true)); + await asyncTest( + () => emptyTest(const Stream.empty(broadcast: false), false)); + + // Check that the behavior is consistent with other empty multi-subscription + // streams. + await asyncTest(() => emptyTest(Stream.fromIterable([]), false)); + await asyncTest(() => + emptyTest((StreamController.broadcast()..close()).stream, true)); + await asyncTest( + () => emptyTest(Stream.multi((c) => c.close()), false)); + }); + await flushMicrotasks(); } -Future runTest() async { - unreachable([a, b]) { - throw "UNREACHABLE"; +// Function which fails the test if it gets called. +void unreachable([a, b]) { + Expect.fail("Unreachable"); +} + +/// Check that something happens once. +class Checker { + bool checked = false; + void check() { + if (checked) Expect.fail("Checked more than once"); + checked = true; } +} - int tick = 0; - ticker() { - tick++; - } +Future emptyTest(Stream s, bool broadcast) async { + var checker = Checker(); + // Respects type parameter (not a `Stream`) + Expect.notType>(s); - asyncStart(); + // Has the expected `isBroadcast`. + Expect.equals(broadcast, s.isBroadcast); - Stream s = const Stream.empty(); // Is const constructor. - Expect.isFalse(s is Stream); // Respects type parameter. StreamSubscription sub = - s.listen(unreachable, onError: unreachable, onDone: ticker); - Expect.isFalse(sub is StreamSubscription); // Type parameter in sub. + s.listen(unreachable, onError: unreachable, onDone: checker.check); + // Type parameter of subscription repspects stream. + // Not a `StreamSubscription`. + Expect.isFalse(sub is StreamSubscription); - // Doesn't do callback in response to listen. - Expect.equals(tick, 0); + Expect.isFalse(sub.isPaused); + // Doesn't do callback immediately in response to listen. + Expect.isFalse(checker.checked); await flushMicrotasks(); - // Completes eventually. - Expect.equals(tick, 1); + // Completes eventually, after a microtask. + Expect.isTrue(checker.checked); + Expect.isFalse(sub.isPaused); - // It's a broadcast stream - can listen twice. - Expect.isTrue(s.isBroadcast); + // Can listen more than once, whether broadcast stream or not. + checker = Checker(); StreamSubscription sub2 = - s.listen(unreachable, onError: unreachable, onDone: unreachable); - // respects pause. + s.listen(unreachable, onError: unreachable, onDone: checker.check); + // Respects pauses. sub2.pause(); - await flushMicrotasks(); - // respects cancel. - sub2.cancel(); - await flushMicrotasks(); - Expect.equals(tick, 1); - // Still not complete. + Expect.isTrue(sub2.isPaused); + sub2.pause(); + Expect.isTrue(sub2.isPaused); + sub2.pause(Future.value(0)); + Expect.isTrue(sub2.isPaused); - StreamSubscription sub3 = - s.listen(unreachable, onError: unreachable, onDone: ticker); - // respects pause. + await flushMicrotasks(); + Expect.isTrue(sub2.isPaused); + Expect.isFalse(checker.checked); + + // Resumes when all pauses resumed. + sub2.resume(); + Expect.isTrue(sub2.isPaused); + await flushMicrotasks(); + Expect.isFalse(checker.checked); + Expect.isTrue(sub2.isPaused); + + sub2.resume(); + Expect.isFalse(sub2.isPaused); + await flushMicrotasks(); + Expect.isTrue(checker.checked); + Expect.isFalse(sub2.isPaused); + + // Can't pause after done. + sub2.pause(Future.value(0)); + Expect.isFalse(sub2.isPaused); + sub2.pause(); + Expect.isFalse(sub2.isPaused); + + // Respects cancel. + var sub3 = s.listen(unreachable, onError: unreachable, onDone: unreachable); + sub3.cancel(); + sub3.onDone(unreachable); + Expect.isFalse(sub3.isPaused); + await flushMicrotasks(); + Expect.isFalse(sub3.isPaused); + + // Can't pause after cancel sub3.pause(); - Expect.equals(tick, 1); - await flushMicrotasks(); - // Doesn't complete while paused. - Expect.equals(tick, 1); - sub3.resume(); - await flushMicrotasks(); - // Now completed. - Expect.equals(tick, 2); + Expect.isFalse(sub3.isPaused); + sub3.pause(Future.value(0)); + Expect.isFalse(sub3.isPaused); + // No errors. - asyncEnd(); + // Respects cancel while paused. + var sub4 = s.listen(unreachable, onError: unreachable, onDone: unreachable); + sub4.pause(); + sub4.cancel(); + sub4.onDone(unreachable); + await flushMicrotasks(); + sub4.resume(); + await flushMicrotasks(); + + // Check that the stream is zone-aware. + // Registers onDone callback. + var log = []; + Zone zone; + + void callback1() { + // Run in correct zone. + Expect.equals(zone, Zone.current); + log.add("don1"); + } + + void callback2() { + // Run in correct zone. + Expect.equals(zone, Zone.current); + log.add("don2"); + } + + zone = Zone.current.fork( + specification: ZoneSpecification(registerCallback: (s, p, z, f) { + if (f == callback1) log.add("reg1"); + if (f == callback2) log.add("reg2"); + return p.registerCallback(z, f); + }, run: (s, p, z, f) { + if (f == callback1) log.add("run1"); + if (f == callback2) log.add("run2"); + return p.run(z, f); + })); + + await zone.run(() async { + var s = Stream.empty(); + + var sub = s.listen(unreachable, onError: unreachable, onDone: callback1); + sub.onDone(callback2); + }); + await flushMicrotasks(); + Expect.listEquals(["reg1", "reg2", "run2", "don2"], log); } Future flushMicrotasks() => new Future.delayed(Duration.zero);