diff --git a/.travis.yml b/.travis.yml
index a9b5166..07e132f 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -7,7 +7,7 @@
   - test: --platform vm
   - test: --platform firefox
   - dartfmt
-  - dartanalyzer
+  - dartanalyzer: --fatal-infos --fatal-warnings .
 
 # Only building master means that we don't run two builds for each pull request.
 branches:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 10727a5..0c3d314 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,7 @@
-## x.x.x
+## 1.4.0
+
+* Add `forEach` to `Pool` to support efficient async processing of an
+  `Iterable`.
 
 * Throw ArgumentError if poolSize <= 0
 
diff --git a/lib/pool.dart b/lib/pool.dart
index a97ec75..779300e 100644
--- a/lib/pool.dart
+++ b/lib/pool.dart
@@ -130,6 +130,103 @@
     }
   }
 
+  /// Returns a [Stream] containing the result of [action] applied to each
+  /// element of [elements].
+  ///
+  /// While [action] is invoked on each element of [elements] in order,
+  /// it's possible the return [Stream] may have items out-of-order – especially
+  /// if the completion time of [action] varies.
+  ///
+  /// If [action] throws an error the source item along with the error object
+  /// and [StackTrace] are passed to [onError], if it is provided. If [onError]
+  /// returns `true`, the error is added to the returned [Stream], otherwise
+  /// it is ignored.
+  ///
+  /// Errors thrown from iterating [elements] will not be passed to
+  /// [onError]. They will always be added to the returned stream as an error.
+  ///
+  /// Note: all of the resources of the this [Pool] will be used when the
+  /// returned [Stream] is listened to until it is completed or canceled.
+  ///
+  /// Note: if this [Pool] is closed before the returned [Stream] is listened
+  /// to, a [StateError] is thrown.
+  Stream<T> forEach<S, T>(
+      Iterable<S> elements, FutureOr<T> Function(S source) action,
+      {bool Function(S item, Object error, StackTrace stack) onError}) {
+    onError ??= (item, e, s) => true;
+
+    var cancelPending = false;
+
+    Completer resumeCompleter;
+    StreamController<T> controller;
+
+    Iterator<S> iterator;
+
+    Future<void> run(int i) async {
+      while (iterator.moveNext()) {
+        // caching `current` is necessary because there are async breaks
+        // in this code and `iterator` is shared across many workers
+        final current = iterator.current;
+
+        _resetTimer();
+
+        await resumeCompleter?.future;
+
+        if (cancelPending) {
+          break;
+        }
+
+        T value;
+        try {
+          value = await action(current);
+        } catch (e, stack) {
+          if (onError(current, e, stack)) {
+            controller.addError(e, stack);
+          }
+          continue;
+        }
+        controller.add(value);
+      }
+    }
+
+    Future doneFuture;
+
+    void onListen() {
+      assert(iterator == null);
+      iterator = elements.iterator;
+
+      assert(doneFuture == null);
+      doneFuture = Future.wait(
+              Iterable<int>.generate(_maxAllocatedResources)
+                  .map((i) => withResource(() => run(i))),
+              eagerError: true)
+          .catchError(controller.addError);
+
+      doneFuture.whenComplete(controller.close);
+    }
+
+    controller = StreamController<T>(
+      sync: true,
+      onListen: onListen,
+      onCancel: () async {
+        assert(!cancelPending);
+        cancelPending = true;
+        await doneFuture;
+      },
+      onPause: () {
+        assert(resumeCompleter == null);
+        resumeCompleter = Completer();
+      },
+      onResume: () {
+        assert(resumeCompleter != null);
+        resumeCompleter.complete();
+        resumeCompleter = null;
+      },
+    );
+
+    return controller.stream;
+  }
+
   /// Closes the pool so that no more resources are requested.
   ///
   /// Existing resource requests remain unchanged.
diff --git a/pubspec.yaml b/pubspec.yaml
index a709842..095c385 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -1,5 +1,5 @@
 name: pool
-version: 1.3.7-dev
+version: 1.4.0-dev
 
 description: A class for managing a finite pool of resources.
 author: Dart Team <misc@dartlang.org>
diff --git a/test/pool_test.dart b/test/pool_test.dart
index 6c9d0a5..d4d7789 100644
--- a/test/pool_test.dart
+++ b/test/pool_test.dart
@@ -27,7 +27,7 @@
         }
         expect(pool.request(), doesNotComplete);
 
-        async.elapse(Duration(seconds: 1));
+        async.elapse(const Duration(seconds: 1));
       });
     });
 
@@ -42,12 +42,12 @@
           // This will only complete once [lastAllocatedResource] is released.
           expect(pool.request(), completes);
 
-          Future.delayed(Duration(microseconds: 1)).then((_) {
+          Future.delayed(const Duration(microseconds: 1)).then((_) {
             lastAllocatedResource.release();
           });
         });
 
-        async.elapse(Duration(seconds: 1));
+        async.elapse(const Duration(seconds: 1));
       });
     });
   });
@@ -68,7 +68,7 @@
         }
         pool.withResource(expectNoAsync());
 
-        async.elapse(Duration(seconds: 1));
+        async.elapse(const Duration(seconds: 1));
       });
     });
 
@@ -86,15 +86,15 @@
           blockedResourceAllocated = true;
         });
 
-        Future.delayed(Duration(microseconds: 1)).then((_) {
+        Future.delayed(const Duration(microseconds: 1)).then((_) {
           expect(blockedResourceAllocated, isFalse);
           completer.complete();
-          return Future.delayed(Duration(microseconds: 1));
+          return Future.delayed(const Duration(microseconds: 1));
         }).then((_) {
           expect(blockedResourceAllocated, isTrue);
         });
 
-        async.elapse(Duration(seconds: 1));
+        async.elapse(const Duration(seconds: 1));
       });
     });
 
@@ -109,18 +109,18 @@
   group("with a timeout", () {
     test("doesn't time out if there are no pending requests", () {
       FakeAsync().run((async) {
-        var pool = Pool(50, timeout: Duration(seconds: 5));
+        var pool = Pool(50, timeout: const Duration(seconds: 5));
         for (var i = 0; i < 50; i++) {
           expect(pool.request(), completes);
         }
 
-        async.elapse(Duration(seconds: 6));
+        async.elapse(const Duration(seconds: 6));
       });
     });
 
     test("resets the timer if a resource is returned", () {
       FakeAsync().run((async) {
-        var pool = Pool(50, timeout: Duration(seconds: 5));
+        var pool = Pool(50, timeout: const Duration(seconds: 5));
         for (var i = 0; i < 49; i++) {
           expect(pool.request(), completes);
         }
@@ -129,41 +129,41 @@
           // This will only complete once [lastAllocatedResource] is released.
           expect(pool.request(), completes);
 
-          Future.delayed(Duration(seconds: 3)).then((_) {
+          Future.delayed(const Duration(seconds: 3)).then((_) {
             lastAllocatedResource.release();
             expect(pool.request(), doesNotComplete);
           });
         });
 
-        async.elapse(Duration(seconds: 6));
+        async.elapse(const Duration(seconds: 6));
       });
     });
 
     test("resets the timer if a resource is requested", () {
       FakeAsync().run((async) {
-        var pool = Pool(50, timeout: Duration(seconds: 5));
+        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.delayed(Duration(seconds: 3)).then((_) {
+        Future.delayed(const Duration(seconds: 3)).then((_) {
           expect(pool.request(), doesNotComplete);
         });
 
-        async.elapse(Duration(seconds: 6));
+        async.elapse(const Duration(seconds: 6));
       });
     });
 
     test("times out if nothing happens", () {
       FakeAsync().run((async) {
-        var pool = Pool(50, timeout: Duration(seconds: 5));
+        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(Duration(seconds: 6));
+        async.elapse(const Duration(seconds: 6));
       });
     });
   });
@@ -437,6 +437,288 @@
     });
   });
 
+  group('forEach', () {
+    Pool pool;
+
+    tearDown(() async {
+      await pool.close();
+    });
+
+    const delayedToStringDuration = Duration(milliseconds: 10);
+
+    Future<String> delayedToString(int i) =>
+        Future.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.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 * 3;
+
+              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();
+      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
+      StreamSubscription subscription;
+
+      subscription = stream.listen(
+        (data) {
+          dataCount++;
+
+          if (int.parse(data) % 3 == 1) {
+            subscription.pause(Future(() async {
+              await Future.delayed(const Duration(milliseconds: 100));
+            }));
+          }
+        },
+        onError: registerException,
+        onDone: expectAsync0(() {
+          expect(dataCount, 40);
+        }),
+      );
+    });
+
+    group('cancel', () {
+      final dataSize = 32;
+      for (var i = 1; i < 5; i++) {
+        test('with pool size $i', () async {
+          pool = Pool(i);
+
+          var stream =
+              pool.forEach(Iterable<int>.generate(dataSize), delayedToString);
+
+          var cancelCompleter = Completer<void>();
+
+          StreamSubscription subscription;
+
+          var eventCount = 0;
+          subscription = stream.listen((data) {
+            eventCount++;
+            if (int.parse(data) == dataSize ~/ 2) {
+              cancelCompleter.complete();
+            }
+          }, onError: registerException);
+
+          await cancelCompleter.future;
+
+          await subscription.cancel();
+
+          expect(eventCount, 1 + dataSize ~/ 2);
+        });
+      }
+    });
+
+    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('error in action, no onError', () async {
+        pool = Pool(20);
+
+        var listFuture = pool.forEach(Iterable<int>.generate(100), (i) async {
+          await Future.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), (i) async {
+          await Future.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);