pkgs/pool/test/pool_test.dart

// Copyright (c) 2014, the Dart project authors.  Please see the AUTHORS file
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.

import 'dart:async';

import 'package:fake_async/fake_async.dart';
import 'package:pool/pool.dart';
import 'package:stack_trace/stack_trace.dart';
import 'package:test/test.dart';

void main() {
  group('request()', () {
    test('resources can be requested freely up to the limit', () {
      var pool = Pool(50);
      for (var i = 0; i < 50; i++) {
        expect(pool.request(), completes);
      }
    });

    test('resources block past the limit', () {
      FakeAsync().run((async) {
        var pool = Pool(50);
        for (var i = 0; i < 50; i++) {
          expect(pool.request(), completes);
        }
        expect(pool.request(), doesNotComplete);

        async.elapse(const Duration(seconds: 1));
      });
    });

    test('a blocked resource is allocated when another is released', () {
      FakeAsync().run((async) {
        var pool = Pool(50);
        for (var i = 0; i < 49; i++) {
          expect(pool.request(), completes);
        }

        pool.request().then((lastAllocatedResource) {
          // This will only complete once [lastAllocatedResource] is released.
          expect(pool.request(), completes);

          Future<void>.delayed(const Duration(microseconds: 1)).then((_) {
            lastAllocatedResource.release();
          });
        });

        async.elapse(const Duration(seconds: 1));
      });
    });
  });

  group('withResource()', () {
    test('can be called freely up to the limit', () {
      var pool = Pool(50);
      for (var i = 0; i < 50; i++) {
        pool.withResource(expectAsync0(() => Completer<void>().future));
      }
    });

    test('blocks the callback past the limit', () {
      FakeAsync().run((async) {
        var pool = Pool(50);
        for (var i = 0; i < 50; i++) {
          pool.withResource(expectAsync0(() => Completer<void>().future));
        }
        pool.withResource(expectNoAsync());

        async.elapse(const Duration(seconds: 1));
      });
    });

    test('a blocked resource is allocated when another is released', () {
      FakeAsync().run((async) {
        var pool = Pool(50);
        for (var i = 0; i < 49; i++) {
          pool.withResource(expectAsync0(() => Completer<void>().future));
        }

        var completer = Completer<void>();
        pool.withResource(() => completer.future);
        var blockedResourceAllocated = false;
        pool.withResource(() {
          blockedResourceAllocated = true;
        });

        Future<void>.delayed(const Duration(microseconds: 1)).then((_) {
          expect(blockedResourceAllocated, isFalse);
          completer.complete();
          return Future<void>.delayed(const Duration(microseconds: 1));
        }).then((_) {
          expect(blockedResourceAllocated, isTrue);
        });

        async.elapse(const Duration(seconds: 1));
      });
    });

    // Regression test for #3.
    test('can be called immediately before close()', () async {
      var pool = Pool(1);
      unawaited(pool.withResource(expectAsync0(() {})));
      await pool.close();
    });
  });

  group('with a timeout', () {
    test("doesn't time out if there are no pending requests", () {
      FakeAsync().run((async) {
        var pool = Pool(50, timeout: const Duration(seconds: 5));
        for (var i = 0; i < 50; i++) {
          expect(pool.request(), completes);
        }

        async.elapse(const Duration(seconds: 6));
      });
    });

    test('resets the timer if a resource is returned', () {
      FakeAsync().run((async) {
        var pool = Pool(50, timeout: const Duration(seconds: 5));
        for (var i = 0; i < 49; i++) {
          expect(pool.request(), completes);
        }

        pool.request().then((lastAllocatedResource) {
          // This will only complete once [lastAllocatedResource] is released.
          expect(pool.request(), completes);

          Future<void>.delayed(const Duration(seconds: 3)).then((_) {
            lastAllocatedResource.release();
            expect(pool.request(), doesNotComplete);
          });
        });

        async.elapse(const Duration(seconds: 6));
      });
    });

    test('resets the timer if a resource is requested', () {
      FakeAsync().run((async) {
        var pool = Pool(50, timeout: const Duration(seconds: 5));
        for (var i = 0; i < 50; i++) {
          expect(pool.request(), completes);
        }
        expect(pool.request(), doesNotComplete);

        Future<void>.delayed(const Duration(seconds: 3)).then((_) {
          expect(pool.request(), doesNotComplete);
        });

        async.elapse(const Duration(seconds: 6));
      });
    });

    test('times out if nothing happens', () {
      FakeAsync().run((async) {
        var pool = Pool(50, timeout: const Duration(seconds: 5));
        for (var i = 0; i < 50; i++) {
          expect(pool.request(), completes);
        }
        expect(pool.request(), throwsA(const TypeMatcher<TimeoutException>()));

        async.elapse(const Duration(seconds: 6));
      });
    });
  });

  group('allowRelease()', () {
    test('runs the callback once the resource limit is exceeded', () async {
      var pool = Pool(50);
      for (var i = 0; i < 49; i++) {
        expect(pool.request(), completes);
      }

      var resource = await pool.request();
      var onReleaseCalled = false;
      resource.allowRelease(() => onReleaseCalled = true);
      await Future<void>.delayed(Duration.zero);
      expect(onReleaseCalled, isFalse);

      expect(pool.request(), completes);
      await Future<void>.delayed(Duration.zero);
      expect(onReleaseCalled, isTrue);
    });

    test('runs the callback immediately if there are blocked requests',
        () async {
      var pool = Pool(1);
      var resource = await pool.request();

      // This will be blocked until [resource.allowRelease] is called.
      expect(pool.request(), completes);

      var onReleaseCalled = false;
      resource.allowRelease(() => onReleaseCalled = true);
      await Future<void>.delayed(Duration.zero);
      expect(onReleaseCalled, isTrue);
    });

    test('blocks the request until the callback completes', () async {
      var pool = Pool(1);
      var resource = await pool.request();

      var requestComplete = false;
      unawaited(pool.request().then((_) => requestComplete = true));

      var completer = Completer<void>();
      resource.allowRelease(() => completer.future);
      await Future<void>.delayed(Duration.zero);
      expect(requestComplete, isFalse);

      completer.complete();
      await Future<void>.delayed(Duration.zero);
      expect(requestComplete, isTrue);
    });

    test('completes requests in request order regardless of callback order',
        () async {
      var pool = Pool(2);
      var resource1 = await pool.request();
      var resource2 = await pool.request();

      var request1Complete = false;
      unawaited(pool.request().then((_) => request1Complete = true));
      var request2Complete = false;
      unawaited(pool.request().then((_) => request2Complete = true));

      var onRelease1Called = false;
      var completer1 = Completer<void>();
      resource1.allowRelease(() {
        onRelease1Called = true;
        return completer1.future;
      });
      await Future<void>.delayed(Duration.zero);
      expect(onRelease1Called, isTrue);

      var onRelease2Called = false;
      var completer2 = Completer<void>();
      resource2.allowRelease(() {
        onRelease2Called = true;
        return completer2.future;
      });
      await Future<void>.delayed(Duration.zero);
      expect(onRelease2Called, isTrue);
      expect(request1Complete, isFalse);
      expect(request2Complete, isFalse);

      // Complete the second resource's onRelease callback first. Even though it
      // was triggered by the second blocking request, it should complete the
      // first one to preserve ordering.
      completer2.complete();
      await Future<void>.delayed(Duration.zero);
      expect(request1Complete, isTrue);
      expect(request2Complete, isFalse);

      completer1.complete();
      await Future<void>.delayed(Duration.zero);
      expect(request1Complete, isTrue);
      expect(request2Complete, isTrue);
    });

    test('runs onRequest in the zone it was created', () async {
      var pool = Pool(1);
      var resource = await pool.request();

      var outerZone = Zone.current;
      runZoned(() {
        var innerZone = Zone.current;
        expect(innerZone, isNot(equals(outerZone)));

        resource.allowRelease(expectAsync0(() {
          expect(Zone.current, equals(innerZone));
        }));
      });

      await pool.request();
    });

    test('request() completes if allowRelease callback throws', () async {
      var pool = Pool(1);
      var resource = await pool.request();

      var requestFuture = pool.request();

      var completer = Completer<void>();
      resource.allowRelease(() => completer.future);

      await Future<void>.delayed(Duration.zero);
      completer.completeError('oh no!');

      await expectLater(requestFuture, completes);
    });

    test('request() does not leak resources when allowRelease throws',
        () async {
      var pool = Pool(1);
      var resource = await pool.request();

      var requestFuture = pool.request();

      var completer = Completer<void>();
      resource.allowRelease(() => completer.future);

      await Future<void>.delayed(Duration.zero);
      completer.completeError('oh no!');

      var resource2 = await requestFuture;
      resource2.release();

      // Without the fix, this will hang because the slot is leaked.
      var nextRequest = pool.request().timeout(
          const Duration(milliseconds: 100),
          onTimeout: () => throw TimeoutException('Leaked!'));

      await expectLater(nextRequest, completes);
    });

    test('throwing in request listener does not corrupt state', () async {
      var pool = Pool(2);
      var resource1 = await pool.request();
      var resource2 = await pool.request();

      var completer1 = Completer<void>();
      resource1.allowRelease(() => completer1.future);
      var completer2 = Completer<void>();
      resource2.allowRelease(() => completer2.future);

      var requestFuture1 = pool.request();
      var requestFuture2 = pool.request();

      var request1Threw = false;
      var requestFuture1WithListener = requestFuture1.then((_) {
        request1Threw = true;
        throw StateError('Listener 1 threw!');
      });

      expect(requestFuture1WithListener, throwsA(isA<StateError>()));

      var request2Completed = false;
      var request2Error = false;
      unawaited(requestFuture2.then((_) {
        request2Completed = true;
      }, onError: (Object e) {
        request2Error = true;
      }));

      await Future<void>.delayed(Duration.zero);

      completer1.complete();
      await Future<void>.delayed(Duration.zero);

      expect(request1Threw, isTrue);
      expect(request2Completed, isFalse);
      expect(request2Error, isFalse);

      completer2.complete();
      await Future<void>.delayed(Duration.zero);

      expect(request2Completed, isTrue);
      expect(request2Error, isFalse);
    });
  });

  group('PoolResource', () {
    test('release() throws StateError if called twice', () async {
      var pool = Pool(1);
      var resource = await pool.request();
      resource.release();
      expect(resource.release, throwsStateError);
    });

    test('allowRelease() throws StateError if called twice', () async {
      var pool = Pool(1);
      var resource = await pool.request();
      resource.allowRelease(() {});
      expect(() => resource.allowRelease(() {}), throwsStateError);
    });

    test('allowRelease() throws if called after release()', () async {
      var pool = Pool(1);
      var resource = await pool.request();
      resource.release();
      expect(() => resource.allowRelease(() {}), throwsStateError);
    });

    test('release() throws if called after allowRelease()', () async {
      var pool = Pool(1);
      var resource = await pool.request();
      resource.allowRelease(() {});
      expect(resource.release, throwsStateError);
    });
  });

  test("done doesn't complete without close", () async {
    var pool = Pool(1);
    unawaited(pool.done.then(expectAsync1((_) {}, count: 0)));

    var resource = await pool.request();
    resource.release();

    await Future<void>.delayed(Duration.zero);
  });

  group('close()', () {
    test('disallows request() and withResource()', () {
      var pool = Pool(1)..close();
      expect(pool.request, throwsStateError);
      expect(() => pool.withResource(() {}), throwsStateError);
    });

    test('can be called multiple times', () async {
      var pool = Pool(1);
      var resource = await pool.request();

      var closeFuture1 = pool.close();
      var closeFuture2 = pool.close();

      expect(closeFuture1, equals(closeFuture2));

      resource.release();
      await closeFuture1;
    });

    test('pending requests are fulfilled', () async {
      var pool = Pool(1);
      var resource1 = await pool.request();
      expect(
          pool.request().then((resource2) {
            resource2.release();
          }),
          completes);
      expect(pool.done, completes);
      expect(pool.close(), completes);
      resource1.release();
    });

    test('pending requests are fulfilled with allowRelease', () async {
      var pool = Pool(1);
      var resource1 = await pool.request();

      var completer = Completer<void>();
      expect(
          pool.request().then((resource2) {
            expect(completer.isCompleted, isTrue);
            resource2.release();
          }),
          completes);
      expect(pool.close(), completes);

      resource1.allowRelease(() => completer.future);
      await Future<void>.delayed(Duration.zero);

      completer.complete();
    });

    test("doesn't complete until all resources are released", () async {
      var pool = Pool(2);
      var resource1 = await pool.request();
      var resource2 = await pool.request();
      var resource3Future = pool.request();

      var resource1Released = false;
      var resource2Released = false;
      var resource3Released = false;
      expect(
          pool.close().then((_) {
            expect(resource1Released, isTrue);
            expect(resource2Released, isTrue);
            expect(resource3Released, isTrue);
          }),
          completes);

      resource1Released = true;
      resource1.release();
      await Future<void>.delayed(Duration.zero);

      resource2Released = true;
      resource2.release();
      await Future<void>.delayed(Duration.zero);

      var resource3 = await resource3Future;
      resource3Released = true;
      resource3.release();
    });

    test('active onReleases complete as usual', () async {
      var pool = Pool(1);
      var resource = await pool.request();

      // Set up an onRelease callback whose completion is controlled by
      // [completer].
      var completer = Completer<void>();
      resource.allowRelease(() => completer.future);
      expect(
          pool.request().then((_) {
            expect(completer.isCompleted, isTrue);
          }),
          completes);

      await Future<void>.delayed(Duration.zero);
      unawaited(pool.close());

      await Future<void>.delayed(Duration.zero);
      completer.complete();
    });

    test('inactive onReleases fire', () async {
      var pool = Pool(2);
      var resource1 = await pool.request();
      var resource2 = await pool.request();

      var completer1 = Completer<void>();
      resource1.allowRelease(() => completer1.future);
      var completer2 = Completer<void>();
      resource2.allowRelease(() => completer2.future);

      expect(
          pool.close().then((_) {
            expect(completer1.isCompleted, isTrue);
            expect(completer2.isCompleted, isTrue);
          }),
          completes);

      await Future<void>.delayed(Duration.zero);
      completer1.complete();

      await Future<void>.delayed(Duration.zero);
      completer2.complete();
    });

    test('new allowReleases fire immediately', () async {
      var pool = Pool(1);
      var resource = await pool.request();

      var completer = Completer<void>();
      expect(
          pool.close().then((_) {
            expect(completer.isCompleted, isTrue);
          }),
          completes);

      await Future<void>.delayed(Duration.zero);
      resource.allowRelease(() => completer.future);

      await Future<void>.delayed(Duration.zero);
      completer.complete();
    });

    test('an onRelease error is piped to the return value', () async {
      var pool = Pool(1);
      var resource = await pool.request();

      var completer = Completer<void>();
      resource.allowRelease(() => completer.future);

      expect(pool.done, throwsA('oh no!'));
      expect(pool.close(), throwsA('oh no!'));

      await Future<void>.delayed(Duration.zero);
      completer.completeError('oh no!');
    });
  });

  group('forEach', () {
    late Pool pool;

    tearDown(() async {
      await pool.close();
    });

    const delayedToStringDuration = Duration(milliseconds: 10);

    Future<String> delayedToString(int i) =>
        Future<String>.delayed(delayedToStringDuration, () => i.toString());

    for (var itemCount in [0, 5]) {
      for (var poolSize in [1, 5, 6]) {
        test('poolSize: $poolSize, itemCount: $itemCount', () async {
          pool = Pool(poolSize);

          var finishedItems = 0;

          await for (var item in pool.forEach(
              Iterable.generate(itemCount, (i) {
                expect(i, lessThanOrEqualTo(finishedItems + poolSize),
                    reason: 'the iterator should be called lazily');
                return i;
              }),
              delayedToString)) {
            expect(int.parse(item), lessThan(itemCount));
            finishedItems++;
          }

          expect(finishedItems, itemCount);
        });
      }
    }

    test('pool closed before listen', () async {
      pool = Pool(2);

      var stream = pool.forEach(Iterable<int>.generate(5), delayedToString);

      await pool.close();

      expect(stream.toList(), throwsStateError);
    });

    test('completes even if the pool is partially used', () async {
      pool = Pool(2);

      var resource = await pool.request();

      var stream = pool.forEach(<int>[], delayedToString);

      expect(await stream.length, 0);

      resource.release();
    });

    test('stream paused longer than timeout', () async {
      pool = Pool(2, timeout: delayedToStringDuration);

      var resource = await pool.request();

      var stream = pool.forEach<int, int>(
          Iterable.generate(100, (i) {
            expect(i, lessThan(20),
                reason: 'The timeout should happen '
                    'before the entire iterable is iterated.');
            return i;
          }), (i) async {
        await Future<void>.delayed(Duration(milliseconds: i));
        return i;
      });

      await expectLater(
          stream.toList,
          throwsA(const TypeMatcher<TimeoutException>().having(
              (te) => te.message,
              'message',
              contains('Pool deadlock: '
                  'all resources have been allocated for too long.'))));

      resource.release();
    });

    group('timing and timeout', () {
      for (var poolSize in [2, 8, 64]) {
        for (var otherTaskCount
            in [0, 1, 7, 63].where((otc) => otc < poolSize)) {
          test('poolSize: $poolSize, otherTaskCount: $otherTaskCount',
              () async {
            final itemCount = 128;
            pool = Pool(poolSize, timeout: const Duration(milliseconds: 20));

            var otherTasks = await Future.wait(
                Iterable<int>.generate(otherTaskCount)
                    .map((i) => pool.request()));

            try {
              var finishedItems = 0;

              var watch = Stopwatch()..start();

              await for (var item in pool.forEach(
                  Iterable.generate(itemCount, (i) {
                    expect(i, lessThanOrEqualTo(finishedItems + poolSize),
                        reason: 'the iterator should be called lazily');
                    return i;
                  }),
                  delayedToString)) {
                expect(int.parse(item), lessThan(itemCount));
                finishedItems++;
              }

              expect(finishedItems, itemCount);

              final expectedElapsed =
                  delayedToStringDuration.inMicroseconds * 4;

              expect((watch.elapsed ~/ itemCount).inMicroseconds,
                  lessThan(expectedElapsed / (poolSize - otherTaskCount)),
                  reason: 'Average time per task should be '
                      'proportionate to the available pool resources.');
            } finally {
              for (var task in otherTasks) {
                task.release();
              }
            }
          });
        }
      }
    }, testOn: 'vm');

    test('partial iteration', () async {
      pool = Pool(5);
      var stream = pool.forEach(Iterable<int>.generate(100), delayedToString);
      expect(await stream.take(10).toList(), hasLength(10));
    });

    test('pool close during data with waiting to be done', () async {
      pool = Pool(5);

      var stream = pool.forEach(Iterable<int>.generate(100), delayedToString);

      var dataCount = 0;
      var subscription = stream.listen((data) {
        dataCount++;
        pool.close();
      });

      await subscription.asFuture<void>();
      expect(dataCount, 100);
      await subscription.cancel();
    });

    test('pause and resume ', () async {
      var generatedCount = 0;
      var dataCount = 0;
      final poolSize = 5;

      pool = Pool(poolSize);

      var stream = pool.forEach(
          Iterable<int>.generate(40, (i) {
            expect(generatedCount, lessThanOrEqualTo(dataCount + 2 * poolSize),
                reason: 'The iterator should not be called '
                    'much faster than the data is consumed.');
            generatedCount++;
            return i;
          }),
          delayedToString);

      // ignore: cancel_subscriptions
      late StreamSubscription subscription;

      subscription = stream.listen(
        (data) {
          dataCount++;

          if (int.parse(data) % 3 == 1) {
            subscription.pause(Future(() async {
              await Future<void>.delayed(const Duration(milliseconds: 100));
            }));
          }
        },
        onError: registerException,
        onDone: expectAsync0(() {
          expect(dataCount, 40);
        }),
      );
    });

    group('cancel stops new work from starting', () {
      final dataSize = 32;
      for (var i = 1; i < 5; i++) {
        test('with pool size $i', () async {
          pool = Pool(i);

          final stopAfter = dataSize ~/ 2;
          var workStarted = -1;
          var stream =
              pool.forEach(Iterable<int>.generate(dataSize), (i) async {
            if (i > workStarted) workStarted = i;
            await pumpEventQueue();
            return i;
          });
          await stream.take(stopAfter).drain<void>();
          expect(workStarted, lessThanOrEqualTo(stopAfter + i),
              reason: 'at most $i resources may be acquired '
                  'during ongoing cancellation');
        });
      }

      test('cancel while paused completes', () {
        FakeAsync().run((async) {
          pool = Pool(2);

          var stream = pool.forEach(Iterable<int>.generate(10), (i) async {
            await Future<void>.delayed(const Duration(milliseconds: 10));
            return i;
          });

          var subscription = stream.listen((_) {});

          async.elapse(const Duration(milliseconds: 5)); // trigger onListen

          subscription.pause();
          async.elapse(const Duration(milliseconds: 50)); // let it pause

          var cancelFuture = subscription.cancel();

          expect(cancelFuture, completes);

          async.elapse(const Duration(seconds: 1)); // let cancel proceed
        });
      });

      test('cancel waits for all workers to finish on error', () async {
        pool = Pool(2);

        var worker1Started = Completer<void>();
        var worker1Finish = Completer<void>();

        var iterable = Iterable.generate(10, (i) {
          if (i == 1) {
            throw StateError('iterator error');
          }
          return i;
        });

        var stream = pool.forEach(iterable, (i) async {
          if (i == 0) {
            worker1Started.complete();
            await worker1Finish.future;
          }
          return i;
        });

        var errorCompleter = Completer<void>();
        var cancelCompleter = Completer<void>();

        late StreamSubscription subscription;

        subscription = stream.listen((data) {}, onError: (e) {
          errorCompleter.complete();
          subscription.cancel().then((_) {
            cancelCompleter.complete();
          });
        });

        await worker1Started.future;
        await errorCompleter.future;

        await pumpEventQueue();
        expect(cancelCompleter.isCompleted, isFalse);

        worker1Finish.complete();
        await cancelCompleter.future;
      });
    });

    group('errors', () {
      Future<void> errorInIterator({
        bool Function(int item, Object error, StackTrace stack)? onError,
      }) async {
        pool = Pool(20);

        var listFuture = pool
            .forEach(
                Iterable.generate(100, (i) {
                  if (i == 50) {
                    throw StateError('error while generating item in iterator');
                  }

                  return i;
                }),
                delayedToString,
                onError: onError)
            .toList();

        await expectLater(() async => listFuture, throwsStateError);
      }

      test('iteration, no onError', () async {
        await errorInIterator();
      });
      test('iteration, with onError', () async {
        await errorInIterator(onError: (i, e, s) => false);
      });

      test('iterator error stops processing other items', () async {
        pool = Pool(2);

        var pulledItems = <int>[];
        var iterable = Iterable.generate(10, (i) {
          pulledItems.add(i);
          if (i == 3) {
            throw StateError('iterator error');
          }
          return i;
        });

        var stream = pool.forEach(iterable, (i) async {
          await Future<void>.delayed(const Duration(milliseconds: 10));
          return i;
        });

        await expectLater(stream.toList(), throwsStateError);

        // Without optimization, it pulls all 10 items.
        // With optimization, it should stop early.
        expect(pulledItems.length, lessThan(10));
      });

      test(
          'iterator error with advancing iterator stops processing other items',
          () async {
        pool = Pool(1);

        var pulledItems = <int>[];
        var stream = pool.forEach(_AdvancingThrowingIterable((i) {
          pulledItems.add(i);
        }), (i) async {
          await Future<void>.delayed(const Duration(milliseconds: 10));
          return i;
        });

        await expectLater(stream.toList(), throwsException);

        expect(pulledItems.length, lessThan(5));
      });

      test('error in action, no onError', () async {
        pool = Pool(20);

        var listFuture = pool.forEach(Iterable<int>.generate(100), (i) async {
          await Future<void>.delayed(const Duration(milliseconds: 10));
          if (i == 10) {
            throw UnsupportedError('10 is not supported');
          }
          return i.toString();
        }).toList();

        await expectLater(() async => listFuture, throwsUnsupportedError);
      });

      test('error in action, no onError', () async {
        pool = Pool(20);

        var list = await pool.forEach(Iterable<int>.generate(100),
            (int i) async {
          await Future<void>.delayed(const Duration(milliseconds: 10));
          if (i % 10 == 0) {
            throw UnsupportedError('Multiples of 10 not supported');
          }
          return i.toString();
        },
            onError: (item, error, stack) =>
                error is! UnsupportedError).toList();

        expect(list, hasLength(90));
      });
    });
  });

  test('throw error when pool limit <= 0', () {
    expect(() => Pool(-1), throwsArgumentError);
    expect(() => Pool(0), throwsArgumentError);
  });
}

/// Returns a function that will cause the test to fail if it's called.
///
/// This should only be called within a [FakeAsync.run] zone.
void Function() expectNoAsync() {
  var stack = Trace.current(1);
  return () => registerException(
      TestFailure('Expected function not to be called.'), stack);
}

/// A matcher for Futures that asserts that they don't complete.
///
/// This should only be called within a [FakeAsync.run] zone.
Matcher get doesNotComplete => predicate((Future future) {
      var stack = Trace.current(1);
      future.then((_) => registerException(
          TestFailure('Expected future not to complete.'), stack));
      return true;
    });

class _AdvancingThrowingIterable extends Iterable<int> {
  final void Function(int) onPull;
  _AdvancingThrowingIterable(this.onPull);

  @override
  Iterator<int> get iterator => _AdvancingThrowingIterator(onPull);
}

class _AdvancingThrowingIterator implements Iterator<int> {
  final void Function(int) onPull;
  int _count = 0;

  _AdvancingThrowingIterator(this.onPull);

  @override
  int get current => _count;

  @override
  bool moveNext() {
    _count++;
    onPull(_count);
    if (_count == 2) {
      throw Exception('Iterator failed');
    }
    return _count < 5;
  }
}