Add Pool.close().

lib/pool.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.
 
 library pool;
 
 import 'dart:async';
 import 'dart:collection';
 
+import 'package:async/async.dart';
 import 'package:stack_trace/stack_trace.dart';
 
 /// Manages an abstract pool of resources with a limit on how many may be in use
 /// at once.
 ///
 /// When a resource is needed, the user should call [request]. When the returned
 /// future completes with a [PoolResource], the resource may be allocated. Once
 /// the resource has been released, the user should call [PoolResource.release].
 /// The pool will ensure that only a certain number of [PoolResource]s may be
 /// allocated at once.
@@ skipping 28 matching lines @@
   /// If [_timeout] isn't null, this timer is set as soon as the resource limit
   /// is reached and is reset every time an resource is released or a new
   /// resource is requested. If it fires, that indicates that the caller became
   /// deadlocked, likely due to files waiting for additional files to be read
   /// before they could be closed.
   Timer _timer;
 
   /// The amount of time to wait before timing out the pending resources.
   final Duration _timeout;
 
+  /// A [FutureGroup] that tracks all the `onRelease` callbacks for resources
+  /// that have been marked releasable.
+  ///
+  /// This is `null` until [close] is called.
+  FutureGroup _closeGroup;
+
+  /// Whether [close] has been called.
+  bool get isClosed => _closeGroup != null;
+
   /// Creates a new pool with the given limit on how many resources may be
   /// allocated at once.
   ///
   /// If [timeout] is passed, then if that much time passes without any activity
   /// all pending [request] futures will throw a [TimeoutException]. This is
   /// intended to avoid deadlocks.
   Pool(this._maxAllocatedResources, {Duration timeout})
       : _timeout = timeout;
 
   /// Request a [PoolResource].
   ///
   /// If the maximum number of resources is already allocated, this will delay
   /// until one of them is released.
   Future<PoolResource> request() {
+    if (isClosed) {
+      throw new StateError("request() may not be called on a closed Pool.");
+    }
+
     if (_allocatedResources < _maxAllocatedResources) {
       _allocatedResources++;
       return new Future.value(new PoolResource._(this));
     } else if (_onReleaseCallbacks.isNotEmpty) {
       return _runOnRelease(_onReleaseCallbacks.removeFirst());
     } else {
       var completer = new Completer<PoolResource>();
       _requestedResources.add(completer);
       _resetTimer();
       return completer.future;
     }
   }
 
   /// Requests a resource for the duration of [callback], which may return a
   /// Future.
   ///
   /// The return value of [callback] is piped to the returned Future.
   Future withResource(callback()) {
-    return request().then((resource) => new Future.sync(callback).whenComplete(resource.release));
+    if (isClosed) {
+      throw new StateError(
+          "withResource() may not be called on a closed Pool.");
+    }
+
+    return request().then((resource) {

[Bob Nystrom, 2015/10/09 00:25:25]

Use async/await?

[nweiz, 2015/10/09 00:46:19]

This is used in test, which means we have to work around
https://github.com/dart-lang/sdk/issues/23497. I'll add a TODO.

+      return new Future.sync(callback).whenComplete(resource.release);
+    });
+  }
+
+  /// Closes the pool so that no more resources are requested.
+  ///
+  /// Existing resource requests remain unchanged.
+  ///
+  /// Any resources that are marked as releasable using
+  /// [PoolResource.allowRelease] are released immediately. Once all resources
+  /// have been released and any `onRelease` callbacks have completed, the
+  /// returned future completes successfully. If any `onRelease` callback throws
+  /// an error, the returned future completes with that error.
+  Future close() {
+    if (_closeGroup != null) return _closeGroup.future;

[Bob Nystrom, 2015/10/09 00:25:25]

Document that it can be called more than once.

[nweiz, 2015/10/09 00:46:19]

Done.

+
+    _resetTimer();
+
+    _closeGroup = new FutureGroup();
+    for (var callback in _onReleaseCallbacks) {
+      _closeGroup.add(new Future.sync(callback));
+    }
+
+    _allocatedResources -= _onReleaseCallbacks.length;
+    _onReleaseCallbacks.clear();
+
+    if (_allocatedResources == 0) _closeGroup.close();
+    return _closeGroup.future;
   }
 
   /// If there are any pending requests, this will fire the oldest one.
   void _onResourceReleased() {
     _resetTimer();
 
-    if (_requestedResources.isEmpty) {
+    if (_requestedResources.isNotEmpty) {
+      var pending = _requestedResources.removeFirst();
+      pending.complete(new PoolResource._(this));
+    } else {
       _allocatedResources--;
-      return;
+      if (isClosed && _allocatedResources == 0) _closeGroup.close();
     }
-
-    var pending = _requestedResources.removeFirst();
-    pending.complete(new PoolResource._(this));
   }
 
   /// If there are any pending requests, this will fire the oldest one after
   /// running [onRelease].
   void _onResourceReleaseAllowed(onRelease()) {
     _resetTimer();
 
-    if (_requestedResources.isEmpty) {
+    if (_requestedResources.isNotEmpty) {
+      var pending = _requestedResources.removeFirst();
+      pending.complete(_runOnRelease(onRelease));
+    } else if (isClosed) {
+      _closeGroup.add(new Future.sync(onRelease));
+      _allocatedResources--;
+      if (_allocatedResources == 0) _closeGroup.close();
+    } else {
       _onReleaseCallbacks.add(
           Zone.current.bindCallback(onRelease, runGuarded: false));
-      return;
     }
-
-    var pending = _requestedResources.removeFirst();
-    pending.complete(_runOnRelease(onRelease));
   }
 
   /// Runs [onRelease] and returns a Future that completes to a resource once an
   /// [onRelease] callback completes.
   ///
   /// Futures returned by [_runOnRelease] always complete in the order they were
   /// created, even if earlier [onRelease] callbacks take longer to run.
   Future<PoolResource> _runOnRelease(onRelease()) {
     new Future.sync(onRelease).then((value) {
       _onReleaseCompleters.removeFirst().complete(new PoolResource._(this));
@@ skipping 66 matching lines @@
   /// produce additional information later on. For example, an isolate's task
   /// may be complete, but it could still emit asynchronous errors.
   void allowRelease(onRelease()) {
     if (_released) {
       throw new StateError("A PoolResource may only be released once.");
     }
     _released = true;
     _pool._onResourceReleaseAllowed(onRelease);
   }
 }
-