diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a86d55..12c98c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +## Unreleased + +* Added `Buffer`, which collects the items passed to it and invokes your + function once with all of them, instead of once per item. +* Added `buffered()` on functions taking a `List`, and a top-level `buffer()`. + ## [1.1.1] - (14-08-2026) * Relaxed the `clock` constraint to `^1.1.1` so packages that depend on diff --git a/README.md b/README.md index bd918f0..a75388d 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ _Rate limiting_ is a strategy for limiting an action. It puts a cap on how often - [Debounce](#debounce) - [Throttle](#throttle) - [BackOff](#backoff) + - [Buffer](#buffer) - [Pending](#pending) - [Flush](#flush) - [Cancellation](#cancellation) @@ -166,11 +167,96 @@ final response = backOff( ); ``` +### Buffer +A _buffered function_ collects the items passed to it and invokes your function **once** with all of them, rather than once per item. Where debounce and throttle keep only the last call's arguments and drop the rest, a buffer keeps every one — so it batches work instead of shedding it. + +The buffer is flushed once `wait` has passed since the first item landed in it, or as soon as it holds `maxSize` items, whichever comes first. Only one flush runs at a time: while your function is working, arriving items collect for the next one, which goes out once it has come due *and* the running one has finished — whichever is later. By default nothing is dropped and no call ever waits on a flush already running — pass `maxQueueSize` to cap the buffer and shed the excess instead. A call that reaches `maxSize` hands its batch over itself, so synchronous work in your function runs before that call returns. + +#### Usage +1. Creating from scratch +```dart +final markRead = buffer((ids) { + print('Marking ${ids.length} messages read'); + return api.markAllRead(ids); +}, const Duration(milliseconds: 500), maxSize: 25); +``` +2. Converting an existing function into a buffered function +```dart +Future markAllRead(List ids) => api.markAllRead(ids); + +final markRead = markAllRead.buffered( + const Duration(milliseconds: 500), + maxSize: 25, +); +``` + +#### Example +Marking messages read as the user scrolls calls `markRead` once per message. Debouncing it would send only the last id and lose the other nineteen; buffering sends one request carrying all twenty. +```dart +void onMessageSeen(String id) { + markRead(id); +} +``` + +Passing `Duration.zero` batches everything queued up in the current event loop turn, which is how a data loader collapses a screen's worth of lookups into one request. +```dart +final loadUsers = buffer( + (ids) => api.getUsers(ids), + Duration.zero, +); +``` + +A producer faster than `onFlush` can grow the buffer without bound. `maxQueueSize` caps it, `overflow` picks which end goes, and `onDrop` reports what that cost. +```dart +final trackEvent = buffer( + (events) => analytics.send(events), + const Duration(seconds: 5), + maxQueueSize: 10000, + overflow: OverflowPolicy.dropOldest, // or dropNewest, to keep what is waiting + onDrop: (events) => log.warning('dropped ${events.length} events'), +); +``` + +The two caps work on different things and compose: `maxSize` limits what any one flush carries, `maxQueueSize` limits the backlog that builds up behind a flush still running. + +Because no caller is waiting on a scheduled flush, failures go to `onError` instead — which is also handed the items, so they can be re-queued rather than lost. +```dart +final markRead = buffer( + (ids) => api.markAllRead(ids), + const Duration(milliseconds: 500), + onError: (error, stackTrace, ids) => retryQueue.addAll(ids), +); +``` + +Handing them back to the buffer retries them on the next flush: +```dart +late final markRead = buffer( + (ids) => api.markAllRead(ids), + const Duration(milliseconds: 500), + onError: (error, stackTrace, ids) => markRead.addAll(ids), +); +``` +Retried items go to the back of the buffer, so their order is not preserved, and nothing spaces the attempts out. For that, put [backOff](#backoff) inside the flush instead: +```dart +final markRead = buffer( + (ids) => backOff( + () => api.markAllRead(ids), + maxAttempts: 4, + retryIf: (error, attempt) => error is SocketException, + ), + const Duration(milliseconds: 500), + // Reached only once backoff has run out of attempts. + onError: (error, stackTrace, ids) => log.warning('gave up on ${ids.length}'), +); +``` +Every attempt re-sends the same batch, `onError` fires once at the end rather than per attempt, and `await markRead.flush()` waits the retries out. Pick one of the two though — requeuing through `onError` *and* retrying with `backOff` compounds the two schedules. + ### Pending Used to check if the there are functions still remaining to get invoked. ```dart final pending = debouncedFunction.isPending; final pending = throttledFunction.isPending; +final pending = bufferedFunction.isPending; ``` ### Flush @@ -178,6 +264,9 @@ Used to immediately invoke all the remaining delayed functions. ```dart final result = debouncedFunction.flush(); final result = throttledFunction.flush(); +// A buffer's flush covers the items it holds right now, and completes once +// your function does, so it can be awaited before going away. +await bufferedFunction.flush(); ``` ### Cancellation @@ -185,4 +274,6 @@ Used to cancel all the remaining delayed functions. ```dart debouncedFunction.cancel(); throttledFunction.cancel(); +// Discards the items collected so far, without invoking your function. +bufferedFunction.cancel(); ``` diff --git a/example/lib/main.dart b/example/lib/main.dart index 91a8488..3117784 100644 --- a/example/lib/main.dart +++ b/example/lib/main.dart @@ -36,4 +36,22 @@ void main() { for (var i = 0; i < 10000; i++) { throttledFunction(); } + + // prints how many values it was handed at once + void printBatchSize(List values) { + print('got ${values.length} values'); + } + + // collects the values it is given and invokes `func` once with all of them, + // instead of once per value + final bufferedFunction = printBatchSize.buffered( + const Duration(milliseconds: 100), + maxSize: 1000, + ); + + // bufferedFunction prints 10 times, once per full buffer, even though + // invoked 10000 times — and no value is dropped + for (var i = 0; i < 10000; i++) { + bufferedFunction(i); + } } diff --git a/lib/rate_limiter.dart b/lib/rate_limiter.dart index 36a3ff8..be1d141 100644 --- a/lib/rate_limiter.dart +++ b/lib/rate_limiter.dart @@ -1,6 +1,7 @@ library rate_limiter; export 'src/backoff.dart'; +export 'src/buffer.dart'; export 'src/debounce.dart'; export 'src/throttle.dart'; export 'src/extension.dart'; diff --git a/lib/src/buffer.dart b/lib/src/buffer.dart new file mode 100644 index 0000000..154bb22 --- /dev/null +++ b/lib/src/buffer.dart @@ -0,0 +1,432 @@ +import 'dart:async'; +import 'dart:collection'; + +import 'package:clock/clock.dart'; + +/// Invoked with every item a [Buffer] collected, in the order they arrived. +typedef BufferFlushCallback = FutureOr Function(List items); + +/// Invoked when a flush a [Buffer] started itself fails, with the items that +/// flush was carrying so they can be re-queued rather than lost. +typedef BufferErrorCallback = void Function( + Object error, + StackTrace stackTrace, + List items, +); + +/// Invoked with the items a [Buffer] dropped to stay within `maxQueueSize`. +typedef BufferDropCallback = void Function(List items); + +/// What a [Buffer] gives up when it is holding `maxQueueSize` items and +/// another one arrives. +enum OverflowPolicy { + /// Drops the items that have been waiting longest, keeping the newest. + dropOldest, + + /// Drops the items that just arrived, keeping the ones already waiting. + dropNewest, +} + +/// Creates a buffered function that collects the items passed to it and +/// invokes `onFlush` **once** with all of them, instead of once per item. +/// +/// The buffer is flushed `wait` after the first item lands in it, or as soon +/// as it holds `maxSize` items, whichever comes first. The buffered function +/// comes with a [Buffer.cancel] method to discard the collected items and a +/// [Buffer.flush] method to invoke `onFlush` immediately. +/// +/// Only one flush runs at a time. Items arriving while `onFlush` is working +/// collect for the next one, which goes out once it has come due — `wait` +/// after its own first item landed, or on reaching `maxSize` — and the running +/// flush has finished, whichever is later. A batch that came due while a flush +/// was running does not then serve its wait a second time. +/// +/// By default nothing is dropped and no call ever waits on a flush already +/// running: this is not a capacity buffer. A call that reaches `maxSize` does +/// hand its batch over itself, so synchronous work in `onFlush` runs before +/// that call returns. `maxSize` caps what any one flush carries, and +/// `maxQueueSize` caps the backlog that builds up behind a slow one, shedding +/// the excess per `overflow`. Where `Debounce` and `Throttle` keep only the +/// arguments of the last call and discard the rest, a [Buffer] keeps them all. +/// +/// Some examples: +/// +/// Collapse a burst of per-item calls into a single request. +/// ```dart +/// final markRead = Buffer( +/// (ids) => api.markAllRead(ids), +/// const Duration(milliseconds: 500), +/// maxSize: 25, +/// ); +/// +/// void onMessageSeen(String id) => markRead(id); +/// ``` +/// +/// Batch everything queued up in the current event loop turn, the way a +/// data loader does, by waiting for no time at all. +/// ```dart +/// final loadUsers = Buffer( +/// (ids) => api.getUsers(ids), +/// Duration.zero, +/// ); +/// ``` +/// +/// Keep a runaway producer from growing the buffer without bound, and log +/// what that costs. +/// ```dart +/// final trackEvent = Buffer( +/// (events) => analytics.send(events), +/// const Duration(seconds: 5), +/// maxQueueSize: 10000, +/// onDrop: (events) => log.warning('dropped ${events.length} events'), +/// ); +/// ``` +/// +/// Send what is left before going away. +/// ```dart +/// Future dispose() => markRead.flush(); +/// ``` +class Buffer { + /// Creates a new instance of [Buffer]. + Buffer( + this._onFlush, + Duration wait, { + int? maxSize, + int? maxQueueSize, + OverflowPolicy overflow = OverflowPolicy.dropOldest, + BufferErrorCallback? onError, + BufferDropCallback? onDrop, + }) : _wait = wait, + _maxSize = _checkPositive(maxSize, 'maxSize'), + _maxQueueSize = _checkPositive(maxQueueSize, 'maxQueueSize'), + _overflow = overflow, + _onError = onError, + _onDrop = onDrop; + + // Checked rather than asserted, because asserts are stripped in release and + // a `maxSize` of zero is worse than a crash there: every buffer looks full + // while each flush takes nothing out of it, so it spins sending empty + // batches and never sends the items it holds. + static int? _checkPositive(int? limit, String name) { + if (limit == null || limit > 0) return limit; + throw ArgumentError.value(limit, name, 'must be greater than zero'); + } + + final BufferFlushCallback _onFlush; + final Duration _wait; + final int? _maxSize; + final int? _maxQueueSize; + final OverflowPolicy _overflow; + final BufferErrorCallback? _onError; + final BufferDropCallback? _onDrop; + + // A queue, so taking a batch off the front and shedding from either end are + // all cheap. Draining N items through a list would move O(N^2) elements, + // since every removal from the front shifts everything behind it. + // + // Each item carries when it arrived and when its batch comes due, which is + // what lets the queue answer both of the awkward questions on its own: + // taking from the front moves the oldest sequence on, shedding from the back + // cannot, and a remainder keeps the deadline of the items it is made of + // rather than being given a fresh one. + final _pending = ListQueue<_Pending>(); + var _nextSeq = 0; + + Timer? _timer; + + // Set when the wait for the oldest item has run out, and cleared only once + // the buffer is empty. A batch leaving does not un-expire the wait of what + // is left behind, because those items are every bit as old. + var _waitIsUp = false; + + // The flush running right now, and the sequence it starts from, so a drain + // can tell whether it is the one carrying what that drain measured. One + // field, because the two can never disagree if there is only one of them. + _Running? _running; + + /// The number of items waiting to get flushed. + /// + /// Counts what is still held. Items handed to `onFlush` are gone from here + /// even while that flush is running. + int get length => _pending.length; + + /// True if there are items waiting to get flushed, or a flush is still + /// running. + bool get isPending => _pending.isNotEmpty || _running != null; + + /// Adds [item] to the buffer, arming the flush if it is the first one in. + void call(T item) => addAll([item]); + + /// Adds every item in [items] to the buffer. + /// + /// With a `maxSize` set, this hands over one full buffer at a time, starting + /// the next only once the one before it has finished. With a `maxQueueSize` + /// set, the excess is dropped as one group rather than an item at a time. + void addAll(Iterable items) { + final dueAt = clock.now().add(_wait); + + // Built before the queue is touched, and before the sequence moves on: + // any iterable can fail part way through, and one that does must not + // leave items behind uncounted and unscheduled. + final incoming = <_Pending>[]; + for (final item in items) { + incoming.add((item: item, seq: _nextSeq + incoming.length, dueAt: dueAt)); + } + + _nextSeq += incoming.length; + _pending.addAll(incoming); + _collect(); + } + + /// Invokes `onFlush` with the items collected so far, and keeps going until + /// everything held when this was called has been handed over. + /// + /// If a flush is already running, this waits for it too, so awaiting the + /// result drains the buffer — which is what makes it safe to call from + /// `dispose`. On a buffer holding nothing it waits out the running flush and + /// no more, leaving items that arrive afterwards to their own window. + /// + /// Whichever call starts a flush owns its failure: one started here lands on + /// the returned future, and one the buffer started itself goes to `onError`. + /// So a drain that waits out a flush already running can complete normally + /// even though that flush failed — `onError` was told instead. + Future flush() { + // Already empty, so the flush running is all there is left to wait for. + // Draining past it would cut short the window of items that have only + // just arrived, and enrol this caller in sending them. + if (_pending.isEmpty) return _running?.settled ?? Future.value(); + + return _drainThrough(_pending.last.seq); + } + + /// Discards the collected items without invoking `onFlush`. + /// + /// A flush already running is left alone; its items were handed over before + /// this was called. + void cancel() { + _timer?.cancel(); + _timer = null; + _waitIsUp = false; + _pending.clear(); + } + + // Hands batches over until the item that arrived at [through] has left the + // buffer, and the flush carrying it has finished. + // + // Measured by sequence rather than by counting batches, because the buffer + // schedules flushes of its own from a completion handler and so gets to the + // queue first: counting only its own batches would leave a drain following + // a steady producer for as long as one kept feeding it. + Future _drainThrough(int through) { + if (_pending.isEmpty || _pending.first.seq > through) { + // Gone from the buffer, but a flush still running may be the one + // carrying it, and this has to outlast that to be worth awaiting. + if (_running case final running? when running.from <= through) { + return running.settled.then((_) => _drainThrough(through)); + } + + // Hands back anything that arrived while this was draining. A batch of + // its own does not pump on the way out, to keep its failures answerable + // here, so without this those items would sit with nothing to move them. + _pump(); + return Future.value(); + } + + if (_running case final running?) { + return running.settled.then((_) => _drainThrough(through)); + } + + return _startFlush(report: false).then((_) => _drainThrough(through)); + } + + // Shared by `call` and `addAll` so a bulk add costs one pass, and so a + // group that overflows is reported to `onDrop` in one piece. + void _collect() { + // Hands off what can go right now, so a group arriving all at once is not + // capped against a batch that was never going to sit in the backlog. + _pump(); + + if (_maxQueueSize case final maxQueueSize? + when _pending.length > maxQueueSize) { + final shed = _shedDownTo(maxQueueSize); + + // The backlog shrank, so whatever is left of it needs its own wait — + // settled before `onDrop`, which is free to throw. + _pump(); + + _onDrop?.call(shed); + } + } + + bool get _isFull { + if (_maxSize case final maxSize?) return _pending.length >= maxSize; + return false; + } + + // Starts a flush if one is due and none is running, otherwise arms the wait. + // Re-entered when a flush settles, so the buffer drains one at a time. + void _pump() { + if (_pending.isEmpty) return; + + // One at a time. Whoever is flushing pumps again on the way out, so a + // batch that came due meanwhile goes then rather than waiting afresh. + if (_running == null && (_waitIsUp || _isFull)) { + // Arms whatever is left over itself, before the callback can run. + _startFlush(report: true); + return; + } + + _armWait(); + } + + // The wait belongs to whoever is at the front, so a head that has left takes + // its wait with it: the timer was armed for its deadline, and a wait that + // ran out for it says nothing about an item that arrived later. + void _rewindWait() { + _timer?.cancel(); + _timer = null; + _waitIsUp = + _pending.isNotEmpty && !clock.now().isBefore(_pending.first.dueAt); + } + + // Waits out the oldest item's deadline, which is carried by the item rather + // than by the batch, so what is left behind is not given a fresh window. + // + // Nothing to arm if the buffer is empty, if a wait is already running, or if + // one has already run out and is only waiting its turn at the queue. + void _armWait() { + if (_pending.isEmpty || _timer != null || _waitIsUp) return; + + final due = _pending.first.dueAt.difference(clock.now()); + _timer = Timer(due, _onDue); + } + + void _onDue() { + _timer = null; + _waitIsUp = true; + _pump(); + } + + // Hands the next batch over and claims the queue while it runs. + // + // The claim is staked before `onFlush` is invoked, because `onFlush` can add + // items synchronously and those have to queue behind this flush rather than + // start a second one. + // + // With `report`, a failure goes to `onError`; without it, the failure is + // left on the returned future for whoever asked for the flush. + Future _startFlush({required bool report}) { + final from = _pending.first.seq; + final items = _take(_maxSize); + + // A completer rather than the flush itself: `onFlush` is invoked + // synchronously below, so there is no future to claim the queue with until + // after the window this is closing. + final settled = Completer(); + _running = (settled: settled.future, from: from); + + // Timed before the callback runs: `onFlush` may work synchronously for a + // while, and a remainder should not be waiting on that work to return. + _armWait(); + + final flushing = report ? _invokeAndReport(items) : _invoke(items); + + void release() { + _running = null; + settled.complete(); + } + + // `ignore` takes the error off this derived future, so a failure cannot + // wedge the queue. + flushing.then( + (_) { + release(); + + // Only the scheduled path pumps on the way out. An explicit flush + // drives its own drain, which is what keeps every batch it sends + // answerable to its caller. + if (report) _pump(); + }, + onError: (Object _, StackTrace __) { + release(); + + // An explicit drain stops at its first failure, so without this the + // items queued behind it would sit with nothing left to move them. + _pump(); + }, + ).ignore(); + + return flushing; + } + + // Sheds down to [maxQueueSize] and hands back what went, for the caller to + // report once scheduling has been settled. + List _shedDownTo(int maxQueueSize) { + final excess = _pending.length - maxQueueSize; + + final List<_Pending> shed; + switch (_overflow) { + case OverflowPolicy.dropOldest: + shed = List.generate(excess, (_) => _pending.removeFirst()); + _rewindWait(); + case OverflowPolicy.dropNewest: + // Off the back, so these arrived last, which is why shedding them can + // never settle a drain measured before they turned up, and why the + // wait at the front is untouched. Handed over in arrival order. + shed = List.generate( + excess, + (_) => _pending.removeLast(), + ).reversed.toList(); + } + + return [for (final pending in shed) pending.item]; + } + + // Takes up to `count` items off the front, disarming the wait. Taken before + // `onFlush` is invoked, so anything added while it runs collects into the + // next batch instead of joining this one. + List _take(int? count) { + final take = + (count == null || count > _pending.length) ? _pending.length : count; + final taken = List.generate(take, (_) => _pending.removeFirst().item); + + _rewindWait(); + + return taken; + } + + // Hands the items over, turning whatever `onFlush` returns into a future so + // the buffer can tell when it is done. + Future _invoke(List items) async => _onFlush(items); + + // A flush the buffer started itself has no caller to hand a failure back to, + // so it goes to `onError`. Without one it is reported to the zone, as an + // unhandled asynchronous error would be. + Future _invokeAndReport(List items) async { + try { + await _invoke(items); + } catch (error, stackTrace) { + final onError = _onError; + if (onError == null) { + Zone.current.handleUncaughtError(error, stackTrace); + return; + } + + try { + onError(error, stackTrace, items); + } catch (handlerError, handlerStackTrace) { + // Reported rather than swallowed: a handler that fails has taken the + // requeue or the logging down with it, which is worse than the flush + // failing in the first place. + Zone.current.handleUncaughtError(handlerError, handlerStackTrace); + } + } + } +} + +// One item, with the sequence it arrived at and the moment its batch is due. +typedef _Pending = ({T item, int seq, DateTime dueAt}); + +// A flush in progress: a future that settles either way, and the sequence of +// the first item it is carrying. +typedef _Running = ({Future settled, int from}); diff --git a/lib/src/extension.dart b/lib/src/extension.dart index d85fc37..e45b855 100644 --- a/lib/src/extension.dart +++ b/lib/src/extension.dart @@ -1,6 +1,7 @@ import 'dart:async'; import 'backoff.dart'; +import 'buffer.dart'; import 'debounce.dart'; import 'throttle.dart'; @@ -24,6 +25,31 @@ extension BackOffExtension on FutureOr Function() { ).call(); } +/// Useful rate limiter extensions for [Function] class. +/// +/// Deliberately not [BufferFlushCallback], which returns `FutureOr`: a +/// plain `void` function is a subtype of this, so both shapes resolve here. +extension BufferExtension on void Function(List items) { + /// Converts this into a [Buffer] function. + Buffer buffered( + Duration wait, { + int? maxSize, + int? maxQueueSize, + OverflowPolicy overflow = OverflowPolicy.dropOldest, + BufferErrorCallback? onError, + BufferDropCallback? onDrop, + }) => + Buffer( + this, + wait, + maxSize: maxSize, + maxQueueSize: maxQueueSize, + overflow: overflow, + onError: onError, + onDrop: onDrop, + ); +} + /// Useful rate limiter extensions for [Function] class. extension RateLimit on Function { /// Converts this into a [Debounce] function. @@ -89,6 +115,26 @@ Debounce debounce( maxWait: maxWait, ); +/// TopLevel lambda to create [Buffer] functions. +Buffer buffer( + BufferFlushCallback onFlush, + Duration wait, { + int? maxSize, + int? maxQueueSize, + OverflowPolicy overflow = OverflowPolicy.dropOldest, + BufferErrorCallback? onError, + BufferDropCallback? onDrop, +}) => + Buffer( + onFlush, + wait, + maxSize: maxSize, + maxQueueSize: maxQueueSize, + overflow: overflow, + onError: onError, + onDrop: onDrop, + ); + /// TopLevel lambda to create [Throttle] functions. Throttle throttle( Function func, diff --git a/test/buffer_backoff_test.dart b/test/buffer_backoff_test.dart new file mode 100644 index 0000000..7227842 --- /dev/null +++ b/test/buffer_backoff_test.dart @@ -0,0 +1,228 @@ +import 'package:fake_async/fake_async.dart'; +import 'package:rate_limiter/rate_limiter.dart'; +import 'package:test/test.dart'; + +import 'utils.dart'; + +class _Unreachable implements Exception {} + +class _Rejected implements Exception {} + +void main() { + group('buffer with a backed-off flush', () { + // randomizationFactor is 0 so the retry delays land where the comments say + // they do: 200ms * 2^attempt, meaning 400ms, then 800ms, then 1600ms. + Buffer build({ + required Future Function(List ids) send, + void Function(Object, StackTrace, List)? onError, + int maxAttempts = 4, + }) { + return buffer( + (ids) => backOff( + () => send(ids), + maxAttempts: maxAttempts, + delayFactor: 200.toDuration(), + randomizationFactor: 0, + retryIf: (error, attempt) => error is _Unreachable, + ), + 500.toDuration(), + onError: onError, + ); + } + + test('should retry the whole batch without telling onError', () { + fakeAsync((async) { + final attempts = >[]; + Object? reported; + + final markRead = build( + send: (ids) async { + attempts.add(ids); + if (attempts.length == 1) throw _Unreachable(); + }, + onError: (e, s, ids) => reported = e, + ); + + markRead('a'); + markRead('b'); + + async.elapse(500.toDuration()); + expect(attempts, [ + ['a', 'b'], + ]); + + async.elapse(400.toDuration()); + expect( + attempts, + [ + ['a', 'b'], + ['a', 'b'], + ], + reason: 'the same batch went out again, intact'); + expect(reported, isNull, reason: 'it succeeded, so nobody was told'); + }); + }); + + test('should hand the items to onError only once backoff gives up', () { + fakeAsync((async) { + var attempts = 0; + Object? reported; + List? handedBack; + + final markRead = build( + maxAttempts: 3, + send: (ids) async { + attempts++; + throw _Unreachable(); + }, + onError: (e, s, ids) { + reported = e; + handedBack = ids; + }, + ); + + markRead('a'); + + async.elapse(500.toDuration()); + expect(reported, isNull, reason: 'still retrying'); + + async.elapse(1200.toDuration()); + expect(attempts, 3); + expect(reported, isA<_Unreachable>()); + expect(handedBack, ['a'], reason: 'so they can be re-queued'); + }); + }); + + test('should go straight to onError for what retryIf rejects', () { + fakeAsync((async) { + var attempts = 0; + Object? reported; + + final markRead = build( + send: (ids) async { + attempts++; + throw _Rejected(); + }, + onError: (e, s, ids) => reported = e, + ); + + markRead('a'); + async.elapse(500.toDuration()); + + expect(attempts, 1); + expect(reported, isA<_Rejected>()); + }); + }); + + test('should hold new items until the retries are done', () { + fakeAsync((async) { + final attempts = >[]; + + final markRead = build( + send: (ids) async { + attempts.add(ids); + if (attempts.length < 3) throw _Unreachable(); + }, + onError: (e, s, ids) {}, + ); + + markRead('a'); + async.elapse(500.toDuration()); // attempt one of 'a' fails + + markRead('b'); + async.elapse(500.toDuration()); // 'b' comes due at t=1000, but waits + + expect( + attempts, + [ + ['a'], + ['a'], + ], + reason: 'only the retry of a has gone out'); + + // The chain runs to t=1700: 400ms then 800ms between its attempts. + async.elapse(700.toDuration()); + + expect( + attempts, + [ + ['a'], + ['a'], + ['a'], + ['b'], + ], + reason: "'b' went the moment the chain freed the queue"); + }); + }); + + test('should keep new items out of a batch being retried', () { + fakeAsync((async) { + final attempts = >[]; + + final markRead = build( + send: (ids) async { + attempts.add(ids); + if (attempts.length == 1) throw _Unreachable(); + }, + onError: (e, s, ids) {}, + ); + + markRead('a'); + async.elapse(500.toDuration()); + + markRead('b'); + async.elapse(400.toDuration()); // the retry of 'a' lands here + + expect(attempts[1], ['a'], + reason: "'b' cannot join a batch already handed over"); + }); + }); + + test('should let flush wait out the retries', () { + fakeAsync((async) { + var attempts = 0; + var drained = false; + + final markRead = build( + send: (ids) async { + attempts++; + if (attempts == 1) throw _Unreachable(); + }, + onError: (e, s, ids) {}, + ); + + markRead('a'); + markRead.flush().then((_) => drained = true).ignore(); + + async.flushMicrotasks(); + expect(drained, isFalse, reason: 'the first attempt has failed'); + + async.elapse(400.toDuration()); + + expect(attempts, 2); + expect(drained, isTrue, reason: 'flush completed once backoff did'); + }); + }); + + test('should report a retry that runs out through flush to its caller', () { + fakeAsync((async) { + Object? caught; + + final markRead = build( + maxAttempts: 2, + send: (ids) async => throw _Unreachable(), + ); + + markRead('a'); + markRead.flush().then((_) {}, onError: (Object e) { + caught = e; + }).ignore(); + + async.elapse(400.toDuration()); + + expect(caught, isA<_Unreachable>(), + reason: 'the caller asked, so the caller is told'); + }); + }); + }); +} diff --git a/test/buffer_test.dart b/test/buffer_test.dart new file mode 100644 index 0000000..903662a --- /dev/null +++ b/test/buffer_test.dart @@ -0,0 +1,1609 @@ +import 'dart:async'; +import 'dart:math'; + +import 'package:fake_async/fake_async.dart'; +import 'package:rate_limiter/rate_limiter.dart'; +import 'package:test/test.dart'; + +import 'utils.dart'; + +void main() { + group('buffer', () { + test('should invoke onFlush once with everything collected', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + buffered('a'); + buffered('b'); + buffered('c'); + + expect(flushes, isEmpty); + expect(buffered.length, 3); + + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b', 'c'], + ]); + }); + }); + + test('should measure the wait from the first item, not the last', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + buffered('a'); + async.elapse(20.toDuration()); + + // A later item rides the deadline already set; it does not push it out + // the way another debounced call would. + buffered('b'); + async.elapse(12.toDuration()); + + expect(flushes, [ + ['a', 'b'], + ]); + }); + }); + + test('should open a new buffer for the items after a flush', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + buffered('a'); + async.elapse(32.toDuration()); + + buffered('b'); + expect(buffered.isPending, isTrue); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a'], + ['b'], + ]); + }); + }); + + test('should flush as soon as it holds maxSize items', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxSize: 2, + ); + + buffered('a'); + buffered('b'); + + // Full, so it went without waiting out the rest of the window. + expect(flushes, [ + ['a', 'b'], + ]); + expect(buffered.isPending, isTrue, + reason: 'the flush is still running'); + + async.flushMicrotasks(); + expect(buffered.isPending, isFalse); + + buffered('c'); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b'], + ['c'], + ]); + }); + }); + + test('should batch a burst from a single event loop turn', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, Duration.zero); + + buffered('a'); + buffered('b'); + buffered('c'); + + expect(flushes, isEmpty, reason: 'the burst has not finished yet'); + + async.elapse(Duration.zero); + + expect(flushes, [ + ['a', 'b', 'c'], + ]); + }); + }); + + test('should fill as many buffers as addAll has items for', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxSize: 2, + ); + + buffered.addAll([1, 2, 3, 4, 5]); + + // One at a time: the second buffer waits for the first to finish. + expect(flushes, [ + [1, 2], + ]); + + async.flushMicrotasks(); + expect(flushes, [ + [1, 2], + [3, 4], + ]); + expect(buffered.length, 1, reason: 'the remainder waits its turn'); + + async.elapse(32.toDuration()); + + expect(flushes, [ + [1, 2], + [3, 4], + [5], + ]); + }); + }); + + test('should fill the buffer already open when addAll arrives', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxSize: 2, + ); + + buffered(1); + buffered.addAll([2, 3, 4]); + async.flushMicrotasks(); + + expect(flushes, [ + [1, 2], + [3, 4], + ]); + expect(buffered.length, 0); + expect(buffered.isPending, isFalse); + }); + }); + + test('should keep collecting after a flush fails', () { + fakeAsync((async) { + final flushes = >[]; + var failNextFlush = true; + + final buffered = buffer( + (items) { + flushes.add(items); + if (failNextFlush) { + failNextFlush = false; + throw Exception('flush failed'); + } + }, + 32.toDuration(), + maxSize: 2, + onError: (e, s, items) {}, + ); + + buffered('a'); + buffered('b'); + + expect(flushes, [ + ['a', 'b'], + ]); + expect(buffered.length, 0); + + buffered('c'); + expect( + buffered.isPending, + isTrue, + reason: 'a failed flush must not wedge the buffer', + ); + + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b'], + ['c'], + ]); + }); + }); + + test('should flush every item when maxSize is one', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxSize: 1, + ); + + buffered('a'); + buffered('b'); + async.flushMicrotasks(); + + expect(flushes, [ + ['a'], + ['b'], + ]); + expect(buffered.isPending, isFalse); + }); + }); + + test('should drop the oldest items once maxQueueSize is reached', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 3, + onDrop: drops.add, + ); + + buffered.addAll(['a', 'b', 'c']); + buffered('d'); + + expect(drops, [ + ['a'], + ]); + expect(buffered.length, 3); + + async.elapse(32.toDuration()); + + expect( + flushes, + [ + ['b', 'c', 'd'], + ], + reason: 'the dropped item never reached onFlush'); + }); + }); + + test('should drop what just arrived when told to keep the oldest', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 3, + overflow: OverflowPolicy.dropNewest, + onDrop: drops.add, + ); + + buffered.addAll(['a', 'b', 'c']); + buffered('d'); + + expect(drops, [ + ['d'], + ]); + + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b', 'c'], + ]); + }); + }); + + test('should report an overflowing group as one drop', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 3, + onDrop: drops.add, + ); + + buffered('a'); + buffered.addAll(['b', 'c', 'd', 'e']); + + // Reported in one piece, rather than once per item over-the-line. + expect(drops, [ + ['a', 'b'], + ]); + + async.elapse(32.toDuration()); + + expect(flushes, [ + ['c', 'd', 'e'], + ]); + }); + }); + + test('should keep the waiting items when dropping the newest', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 2, + overflow: OverflowPolicy.dropNewest, + onDrop: drops.add, + ); + + buffered('a'); + buffered.addAll(['b', 'c', 'd']); + + // 'a' has been waiting, so it survives an incoming group too big to + // fit alongside it. + expect(drops, [ + ['c', 'd'], + ]); + + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b'], + ]); + }); + }); + + test('should keep only the newest when a group dwarfs maxQueueSize', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 2, + onDrop: drops.add, + ); + + buffered.addAll([1, 2, 3, 4, 5]); + + expect(drops, [ + [1, 2, 3], + ]); + expect(buffered.length, 2); + + async.elapse(32.toDuration()); + + expect(flushes, [ + [4, 5], + ]); + }); + }); + + test('should drop without an onDrop to report to', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 2, + ); + + buffered.addAll(['a', 'b', 'c']); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['b', 'c'], + ]); + }); + }); + + test('should discard the collected items on cancel', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + buffered('a'); + buffered('b'); + buffered.cancel(); + + expect(buffered.length, 0); + expect(buffered.isPending, isFalse); + + // Past the deadline, so `cancel` is what stopped it. + async.elapse(128.toDuration()); + + expect(flushes, isEmpty); + }); + }); + + test('should never have two flushes running at once', () { + fakeAsync((async) { + final started = >[]; + final blockers = >[]; + var running = 0; + var mostAtOnce = 0; + + final buffered = buffer( + (items) { + started.add(items); + running++; + mostAtOnce = running > mostAtOnce ? running : mostAtOnce; + final blocker = Completer(); + blockers.add(blocker); + return blocker.future.whenComplete(() => running--); + }, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); + + buffered('b'); + async.elapse(32.toDuration()); + + buffered('c'); + async.elapse(32.toDuration()); + + expect( + started, + [ + ['a'], + ], + reason: "'b' and 'c' waited behind the flush carrying 'a'"); + expect(mostAtOnce, 1); + expect(buffered.isPending, isTrue); + + for (final blocker in blockers.toList()) { + blocker.complete(); + } + async.elapse(32.toDuration()); + + expect( + started, + [ + ['a'], + ['b', 'c'], + ], + reason: 'the two that waited went together'); + expect(mostAtOnce, 1); + }); + }); + + test('should not let onFlush start a second flush from inside itself', () { + fakeAsync((async) { + final started = >[]; + var running = 0; + var mostAtOnce = 0; + var reentered = false; + + late final Buffer buffered; + buffered = buffer( + (items) { + started.add(items); + running++; + mostAtOnce = running > mostAtOnce ? running : mostAtOnce; + + if (!reentered) { + reentered = true; + // Re-entrant, and maxSize of one makes it due immediately. The + // claim on the queue has to be staked before we get here. + buffered('b'); + } + + return Future.delayed(10.toDuration(), () => running--); + }, + 32.toDuration(), + maxSize: 1, + ); + + buffered('a'); + + expect( + started, + [ + ['a'], + ], + reason: "'b' has to wait its turn"); + expect(mostAtOnce, 1); + + async.elapse(100.toDuration()); + + expect(started, [ + ['a'], + ['b'], + ]); + expect(mostAtOnce, 1); + }); + }); + + test('should hold a remainder to its own deadline, not the flush ahead', + () { + fakeAsync((async) { + final blocker = Completer(); + final flushes = >[]; + + final buffered = buffer( + (items) { + flushes.add(items); + return flushes.length == 1 ? blocker.future : null; + }, + 500.toDuration(), + maxSize: 2, + ); + + // [1, 2] fills a batch and goes; 3 stays behind with its own window + // running from now. + buffered.addAll([1, 2, 3]); + + expect(flushes, [ + [1, 2], + ]); + + // The batch ahead of it takes far longer than that window. + async.elapse(2000.toDuration()); + expect(flushes.length, 1, reason: 'still waiting for the queue'); + + blocker.complete(); + async.flushMicrotasks(); + + expect( + flushes, + [ + [1, 2], + [3], + ], + reason: '3 was overdue, so it went as soon as its turn came'); + }); + }); + + test('should go straight out when a wait was served during a flush', () { + fakeAsync((async) { + final started = >[]; + final blocker = Completer(); + + final buffered = buffer( + (items) { + started.add(items); + return started.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); // flush of 'a' starts and blocks + + buffered('b'); + async.elapse(32.toDuration()); // 'b' comes due, but has to wait + + expect(started, [ + ['a'], + ]); + + blocker.complete(); + async.flushMicrotasks(); + + // No second wait: 'b' was already overdue when the flush finished. + expect(started, [ + ['a'], + ['b'], + ]); + }); + }); + + test('should report itself as pending while a flush is running', () { + fakeAsync((async) { + final blocker = Completer(); + + final buffered = buffer( + (items) => blocker.future, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); + + expect(buffered.length, 0, reason: 'handed over'); + expect(buffered.isPending, isTrue, reason: 'but not done'); + + blocker.complete(); + async.flushMicrotasks(); + + expect(buffered.isPending, isFalse); + }); + }); + + test('should let flush await the one already running', () { + fakeAsync((async) { + final started = >[]; + final blocker = Completer(); + var drained = false; + + final buffered = buffer( + (items) { + started.add(items); + return started.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); // flush of 'a' starts and blocks + + buffered('b'); + buffered.flush().then((_) => drained = true).ignore(); + async.flushMicrotasks(); + + expect(drained, isFalse, reason: 'the flush carrying a is still going'); + expect(started, [ + ['a'], + ]); + + blocker.complete(); + async.flushMicrotasks(); + + expect(started, [ + ['a'], + ['b'], + ]); + expect(drained, isTrue, reason: 'and everything after it went too'); + }); + }); + + test('should drain every full buffer on flush', () { + fakeAsync((async) { + final flushes = >[]; + var drained = false; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxSize: 2, + ); + + buffered.addAll([1, 2, 3, 4, 5]); + buffered.flush().then((_) => drained = true).ignore(); + async.flushMicrotasks(); + + expect( + flushes, + [ + [1, 2], + [3, 4], + [5], + ], + reason: 'flush keeps going until nothing is held'); + expect(drained, isTrue); + expect(buffered.isPending, isFalse); + }); + }); + + test('should keep the queue moving when a flush fails', () { + fakeAsync((async) { + final started = >[]; + final blocker = Completer(); + + final buffered = buffer( + (items) { + started.add(items); + return started.length == 1 + ? blocker.future.then((_) => throw Exception('flush failed')) + : null; + }, + 32.toDuration(), + onError: (e, s, items) {}, + ); + + buffered('a'); + async.elapse(32.toDuration()); + + buffered('b'); + blocker.complete(); + async.elapse(32.toDuration()); + + expect( + started, + [ + ['a'], + ['b'], + ], + reason: 'a failure must not wedge what is queued behind it'); + expect(buffered.isPending, isFalse); + }); + }); + + test('should keep the queue moving when an explicit flush fails', () { + fakeAsync((async) { + final blocker = Completer(); + final flushes = >[]; + + final buffered = buffer( + (items) { + flushes.add(items); + return flushes.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + maxSize: 2, + onError: (e, s, items) {}, + ); + + buffered('a'); + buffered.flush().then((_) {}, onError: (Object _) {}).ignore(); + + // One call, so this goes straight to the full branch and never arms a + // wait of its own. The failed drain is all that could move it. + buffered.addAll(['b', 'c']); + + blocker.completeError(Exception('flush failed')); + async.flushMicrotasks(); + + expect( + flushes, + [ + ['a'], + ['b', 'c'], + ], + reason: 'a failed drain must not strand what queued behind it'); + expect(buffered.length, 0); + expect(buffered.isPending, isFalse); + }); + }); + + test('should send what it can before capping the backlog', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + final blocker = Completer(); + + final buffered = buffer( + (items) { + flushes.add(items); + return blocker.future; + }, + 32.toDuration(), + maxSize: 2, + maxQueueSize: 3, + onDrop: drops.add, + ); + + buffered.addAll([1, 2, 3, 4, 5, 6]); + + // Nothing was running, so [1, 2] takes the batch and only what is + // genuinely backlog gets measured against maxQueueSize. + expect(flushes, [ + [1, 2], + ]); + expect(drops, [ + [3], + ]); + expect(buffered.length, 3); + }); + }); + + test('should wait out the running flush and no more when empty', () { + fakeAsync((async) { + final blocker = Completer(); + final flushes = >[]; + var drained = false; + + final buffered = buffer( + (items) { + flushes.add(items); + return flushes.length == 1 ? blocker.future : null; + }, + 1000.toDuration(), + ); + + buffered(1); + async.elapse(1000.toDuration()); // the flush of 1 starts and blocks + expect(buffered.length, 0, reason: 'handed over already'); + + buffered.flush().then((_) => drained = true).ignore(); + + // Arrives after the drain was asked for, with its window still to run. + buffered(2); + + blocker.complete(); + async.flushMicrotasks(); + + expect(drained, isTrue); + expect( + flushes, + [ + [1], + ], + reason: 'the window of 2 was not cut short by an unrelated flush'); + expect(buffered.length, 1); + }); + }); + + test('should outlast the flush that carries what it measured', () { + fakeAsync((async) { + final first = Completer(); + final second = Completer(); + final started = >[]; + var drained = false; + + final buffered = buffer( + (items) { + started.add(items); + return started.length == 1 ? first.future : second.future; + }, + 32.toDuration(), + maxSize: 2, + ); + + buffered.addAll(['a', 'b']); // goes at once, and blocks + buffered.addAll(['c', 'd']); // the snapshot + buffered.flush().then((_) => drained = true).ignore(); + + // Its completion handler starts [c, d], so the snapshot has left the + // buffer — but into a flush that has not finished. + first.complete(); + async.flushMicrotasks(); + + expect(started, [ + ['a', 'b'], + ['c', 'd'], + ]); + expect(drained, isFalse, reason: 'awaiting it must mean it is sent'); + + second.complete(); + async.flushMicrotasks(); + + expect(drained, isTrue); + }); + }); + + test('should not let a shed item take the wait of what replaced it', () { + fakeAsync((async) { + final sentAt = []; + + final buffered = buffer( + (items) => sentAt.add(async.elapsed.inMilliseconds), + 32.toDuration(), + maxQueueSize: 2, + ); + + buffered('a'); // due at t=32 + async.elapse(16.toDuration()); + + // Sheds 'a', so the wait armed for it belongs to nobody. What is left + // arrived at t=16 and is due at t=48. + buffered.addAll(['b', 'c']); + async.elapse(200.toDuration()); + + expect(sentAt, [48]); + }); + }); + + test('should not hand a later item an expired wait when batching', () { + fakeAsync((async) { + final blocker = Completer(); + final sentAt = []; + + final buffered = buffer( + (items) { + sentAt.add(async.elapsed.inMilliseconds); + return sentAt.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + maxSize: 2, + ); + + buffered(0); + async.elapse(32.toDuration()); // sends [0], and blocks + + buffered(1); // due at t=64 + async.elapse(40.toDuration()); // t=72, so 1 is overdue + + buffered.addAll([2, 3]); // due at t=104 + blocker.complete(); + async.elapse(500.toDuration()); + + // [1, 2] goes as soon as the queue frees, since 1 was overdue. 3 was + // not, and must not inherit the wait that ran out for 1. + expect(sentAt, [32, 72, 104]); + }); + }); + + test('should keep a remainder to its deadline across full batches', () { + fakeAsync((async) { + final blocker = Completer(); + final sentAt = []; + + final buffered = buffer( + (items) { + sentAt.add(async.elapsed.inMilliseconds); + return sentAt.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + maxSize: 2, + ); + + // All five arrive together, so all five are due at t=32. + buffered.addAll([1, 2, 3, 4, 5]); + async.elapse(100.toDuration()); + + blocker.complete(); + async.elapse(500.toDuration()); + + // The odd one out is overdue by the time its turn comes, so it goes + // then rather than starting a window of its own. + expect(sentAt, [0, 100, 100]); + }); + }); + + test('should leave the buffer alone when an iterable fails part way', () { + fakeAsync((async) { + final flushes = >[]; + final buffered = buffer(flushes.add, 32.toDuration()); + + expect( + () => buffered.addAll(_failsAfter(3)), + throwsA(isA()), + ); + + expect(buffered.length, 0, reason: 'nothing half-added'); + expect(buffered.isPending, isFalse); + + async.elapse(128.toDuration()); + expect(flushes, isEmpty); + }); + }); + + test('should not let shedding the newest satisfy a drain', () { + fakeAsync((async) { + final blocker = Completer(); + final sent = >[]; + var drained = false; + + final buffered = buffer( + (items) { + sent.add(items); + return sent.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + maxQueueSize: 2, + overflow: OverflowPolicy.dropNewest, + ); + + buffered('z'); + async.elapse(32.toDuration()); // the flush of z starts, and blocks + + buffered.addAll(['a', 'b']); + buffered.flush().then((_) => drained = true).ignore(); + + // Over the cap, so these are shed. They arrived after the drain took + // its measure, so shedding them settles nothing it was waiting for. + buffered.addAll(['c', 'd']); + + blocker.complete(); + async.flushMicrotasks(); + + expect(sent, [ + ['z'], + ['a', 'b'], + ]); + expect(drained, isTrue, reason: 'and only once its own items went'); + expect(buffered.length, 0); + }); + }); + + test('should hand back what came due behind a slow drain', () { + fakeAsync((async) { + final blocker = Completer(); + final sent = >[]; + + final buffered = buffer( + (items) { + sent.add(items); + return sent.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + ); + + buffered('a'); + buffered.flush().ignore(); // takes a, and blocks + + buffered('b'); + async.elapse(50.toDuration()); // b comes due while the drain runs + + blocker.complete(); + async.flushMicrotasks(); + + // The drain is finished, so what it was not measuring has to go back + // under the buffer's own ownership rather than sit unowned. + expect(sent, [ + ['a'], + ['b'], + ]); + expect(buffered.length, 0); + expect(buffered.isPending, isFalse); + }); + }); + + test('should not follow a producer past the batch it measured', () { + fakeAsync((async) { + final blocker = Completer(); + final sent = >[]; + var drained = false; + + final buffered = buffer( + (items) { + sent.add(items); + return sent.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + maxSize: 2, + ); + + buffered.addAll(['a', 'b']); // goes at once, and blocks + buffered.addAll(['c', 'd']); // what the drain takes the measure of + + buffered.flush().then((_) => drained = true).ignore(); + + // Arrives after that measure, and on its own is not a full batch, so + // only a drain that overreached would send it. + buffered('e'); + + blocker.complete(); + async.flushMicrotasks(); + + // The buffer schedules [c, d] itself from the completion handler, so + // the drain never sends a batch of its own — it is done all the same. + expect(sent, [ + ['a', 'b'], + ['c', 'd'], + ]); + expect(drained, isTrue, reason: 'its batch is gone, whoever sent it'); + expect(buffered.length, 1, reason: 'e is the buffer\'s own work'); + }); + }); + + test('should not enrol a flush in what arrives mid-drain', () { + fakeAsync((async) { + final blocker = Completer(); + var calls = 0; + Object? caughtByFlush; + Object? caughtByOnError; + var drained = false; + + final buffered = buffer( + (items) { + calls++; + if (calls == 1) return blocker.future; + throw Exception('the later batch failed'); + }, + 32.toDuration(), + maxSize: 2, + onError: (e, s, items) => caughtByOnError = e, + ); + + buffered('a'); + buffered.flush().then((_) => drained = true, onError: (Object e) { + caughtByFlush = e; + }).ignore(); + + // Arriving after the drain took its measure, so these are the + // buffer's own work rather than the caller's. + buffered('b'); + buffered('c'); + + blocker.complete(); + async.elapse(32.toDuration()); + + expect(calls, 2); + expect(drained, isTrue, reason: 'the one item it took went out'); + expect(caughtByFlush, isNull, reason: 'the caller did not send those'); + expect(caughtByOnError, isNotNull, reason: 'so the buffer answers'); + }); + }); + + test('should leave a flush it only waited for to onError', () { + fakeAsync((async) { + var calls = 0; + Object? caughtByFlush; + Object? caughtByOnError; + var drained = false; + + final buffered = buffer( + (items) { + calls++; + if (calls == 2) throw Exception('the second chunk failed'); + }, + 32.toDuration(), + maxSize: 2, + onError: (e, s, items) => caughtByOnError = e, + ); + + // Two full buffers, so the buffer schedules both chunks itself and + // flush only waits them out. + buffered.addAll([1, 2, 3, 4]); + buffered.flush().then((_) { + drained = true; + }, onError: (Object e) { + caughtByFlush = e; + }).ignore(); + + async.elapse(32.toDuration()); + + expect(calls, 2); + expect(caughtByOnError, isNotNull, reason: 'the buffer sent it'); + expect(caughtByFlush, isNull); + expect(drained, isTrue, + reason: 'the buffer did drain, and it is empty'); + expect(buffered.length, 0); + }); + }); + + test('should collect again after being cancelled', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + expect(buffered.isPending, isFalse); + buffered.cancel(); + + buffered('a'); + buffered.cancel(); + + buffered('b'); + async.elapse(32.toDuration()); + + expect( + flushes, + [ + ['b'], + ], + reason: 'cancel drops what is held, it does not stop the buffer'); + }); + }); + + test('should leave a flush already on its way out alone on cancel', () { + fakeAsync((async) { + final flushes = >[]; + final completer = Completer(); + + final buffered = buffer( + (items) { + flushes.add(items); + return completer.future; + }, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); + + buffered('b'); + buffered.cancel(); + completer.complete(); + async.elapse(32.toDuration()); + + expect( + flushes, + [ + ['a'], + ], + reason: 'the in-flight items were already handed over'); + }); + }); + + test('should invoke onFlush immediately on flush', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + buffered('a'); + buffered.flush().ignore(); + async.flushMicrotasks(); + + expect(flushes, [ + ['a'], + ]); + expect(buffered.isPending, isFalse); + + // The timer it left behind would flush an empty buffer, or worse, + // report itself as pending again. + async.elapse(128.toDuration()); + + expect(flushes, [ + ['a'], + ]); + }); + }); + + test('should complete flush once onFlush does', () { + fakeAsync((async) { + final completer = Completer(); + var flushed = false; + + final buffered = buffer( + (items) => completer.future, + 32.toDuration(), + ); + + buffered('a'); + buffered.flush().then((_) => flushed = true).ignore(); + async.elapse(128.toDuration()); + + expect(flushed, isFalse, reason: 'onFlush has not finished'); + + completer.complete(); + async.flushMicrotasks(); + + expect(flushed, isTrue); + }); + }); + + test('should do nothing when flushing an empty buffer', () { + fakeAsync((async) { + var callCount = 0; + + final buffered = + buffer((items) => callCount++, 32.toDuration()); + + buffered.flush().ignore(); + buffered.addAll(const []); + async.elapse(128.toDuration()); + + expect(callCount, 0); + expect(buffered.isPending, isFalse); + }); + }); + + test('should collect what arrives while onFlush is running', () { + fakeAsync((async) { + final flushes = >[]; + final completer = Completer(); + + final buffered = buffer( + (items) { + flushes.add(items); + return completer.future; + }, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); + + // Joining the buffer being flushed would hand these to a request + // already on its way out. + buffered('b'); + completer.complete(); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a'], + ['b'], + ]); + }); + }); + + test('should hand a scheduled failure to onError', () { + fakeAsync((async) { + final error = Exception('flush failed'); + Object? caughtError; + List? caughtItems; + + final buffered = buffer( + (items) => throw error, + 32.toDuration(), + onError: (e, s, items) { + caughtError = e; + caughtItems = items; + }, + ); + + buffered('a'); + buffered('b'); + async.elapse(32.toDuration()); + + expect(caughtError, error); + // Handed back so they can be re-queued rather than lost. + expect(caughtItems, ['a', 'b']); + }); + }); + + test('should give a failure to the caller that asked for the flush', () { + fakeAsync((async) { + final error = Exception('flush failed'); + Object? caughtByFlush; + Object? caughtByOnError; + + final buffered = buffer( + (items) async => throw error, + 32.toDuration(), + onError: (e, s, items) => caughtByOnError = e, + ); + + buffered('a'); + buffered.flush().onError((e, s) => caughtByFlush = e); + async.flushMicrotasks(); + + expect(caughtByFlush, error); + expect( + caughtByOnError, + isNull, + reason: 'the caller was told, so onError has nothing to report', + ); + }); + }); + + test('should report a scheduled failure with no onError to the zone', () { + final error = Exception('flush failed'); + final caught = Completer(); + + return runZonedGuarded(() async { + final buffered = buffer( + (items) async => throw error, + Duration.zero, + ); + + buffered('a'); + + expect(await caught.future, error); + }, (e, s) { + if (!caught.isCompleted) caught.complete(e); + }); + }); + + test('should report an onError that fails itself to the zone', () { + final caught = Completer(); + + return runZonedGuarded(() async { + final buffered = buffer( + (items) async => throw Exception('flush failed'), + Duration.zero, + // Takes the requeue down with it, so it cannot be swallowed. + onError: (e, s, items) => throw StateError('the handler failed too'), + ); + + buffered('a'); + + expect(await caught.future, isA()); + }, (e, s) { + if (!caught.isCompleted) caught.complete(e); + }); + }); + + test('should convert an existing function into a buffered one', () { + fakeAsync((async) { + final flushes = >[]; + + void markAllRead(List ids) => flushes.add(ids); + + final buffered = markAllRead.buffered(32.toDuration(), maxSize: 3); + + buffered('a'); + buffered('b'); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b'], + ]); + }); + }); + + test('should carry every option through the extension', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + void markAllRead(List ids) => flushes.add(ids); + + final buffered = markAllRead.buffered( + 32.toDuration(), + maxQueueSize: 2, + overflow: OverflowPolicy.dropNewest, + onDrop: drops.add, + ); + + buffered.addAll(['a', 'b', 'c']); + async.elapse(32.toDuration()); + + expect(drops, [ + ['c'], + ]); + expect(flushes, [ + ['a', 'b'], + ]); + }); + }); + + test('should hold its invariants whatever order it is driven in', () { + // What this is really for: whatever order the buffer is driven in, it + // drains once you stop feeding it, and it never hands over more than it + // was told to. Asserting `isPending` here would prove nothing, since it + // reads the queue and so is true by definition wherever items are held. + for (var seed = 0; seed < 200; seed++) { + final rng = Random(seed); + final byMaxSize = rng.nextBool(); + final limit = 1 + rng.nextInt(4); + + // Half the seeds flush slowly, so items come due while a flush is + // still running. Every strand bug so far has hidden in that window. + final slowFlush = rng.nextBool(); + + fakeAsync((async) { + final flushes = >[]; + final buffered = Buffer( + (items) { + flushes.add(items); + return slowFlush ? Future.delayed(25.toDuration()) : null; + }, + 10.toDuration(), + maxSize: byMaxSize ? limit : null, + maxQueueSize: byMaxSize ? null : limit, + ); + + void checkInvariants(String op) { + final where = 'seed $seed, after $op'; + + // `maxSize` bounds each flush, not the buffer: items pile up + // behind a running flush. Only `maxQueueSize` caps what is held. + if (!byMaxSize) { + expect( + buffered.length, + lessThanOrEqualTo(limit), + reason: '$where: held more than maxQueueSize allows', + ); + } + } + + var next = 0; + for (var step = 0; step < 30; step++) { + switch (rng.nextInt(5)) { + case 0: + buffered(next++); + checkInvariants('call'); + case 1: + buffered.addAll(List.generate(rng.nextInt(6), (_) => next++)); + checkInvariants('addAll'); + case 2: + buffered.flush().ignore(); + checkInvariants('flush'); + case 3: + buffered.cancel(); + checkInvariants('cancel'); + case 4: + async.elapse(rng.nextInt(20).toDuration()); + checkInvariants('elapse'); + } + } + + // Nothing more goes in, so everything held has to find its way out. + // A buffer that never armed a timer, or that dropped a wait when the + // head moved, is stuck here rather than merely late. + async.elapse(const Duration(minutes: 1)); + + expect( + buffered.length, + 0, + reason: 'seed $seed: stopped feeding it and it never drained', + ); + expect(buffered.isPending, isFalse, reason: 'seed $seed'); + + for (final flush in flushes) { + expect(flush, isNotEmpty, reason: 'seed $seed: flushed nothing'); + if (byMaxSize) { + expect( + flush.length, + lessThanOrEqualTo(limit), + reason: 'seed $seed: handed over more than maxSize', + ); + } + } + }); + } + }); + + // Thrown rather than asserted: asserts are stripped in release, and a + // maxSize of zero leaves the buffer spinning on empty flushes there. + test('should reject a maxSize that can never fill', () { + expect( + () => Buffer((items) {}, Duration.zero, maxSize: 0), + throwsA(isA()), + ); + }); + + test('should reject a maxQueueSize that can never hold anything', () { + expect( + () => Buffer((items) {}, Duration.zero, maxQueueSize: -1), + throwsA(isA()), + ); + }); + + test('should cap the batch and the backlog independently', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + final blocker = Completer(); + + // maxSize caps what one flush carries; maxQueueSize caps what may pile + // up behind it. Serialized flushes are what make both reachable. + final buffered = buffer( + (items) { + flushes.add(items); + return blocker.future; + }, + 32.toDuration(), + maxSize: 2, + maxQueueSize: 3, + onDrop: drops.add, + ); + + buffered.addAll([1, 2]); + + expect( + flushes, + [ + [1, 2], + ], + reason: 'full, so one flush carrying maxSize items', + ); + + // These pile up behind the running flush, and that backlog is what + // maxQueueSize caps. + buffered.addAll([3, 4, 5, 6]); + + expect(buffered.length, 3); + expect(drops, [ + [3], + ]); + + blocker.complete(); + async.elapse(32.toDuration()); + + expect(flushes, [ + [1, 2], + [4, 5], + [6], + ]); + }); + }); + + test('should take an asynchronous onFlush in either form', () { + fakeAsync((async) { + final flushes = >[]; + + Future markAllRead(List ids) async => flushes.add(ids); + + final fromExtension = markAllRead.buffered(32.toDuration()); + final fromLambda = buffer(markAllRead, 32.toDuration()); + + fromExtension('a'); + fromLambda('b'); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a'], + ['b'], + ]); + }); + }); + }); +} + +// Yields [count] values and then fails, the way a lazy source backed by a +// stream or a database cursor can. +Iterable _failsAfter(int count) sync* { + for (var i = 0; i < count; i++) { + yield i; + } + throw StateError('iteration failed'); +}