added readAsString to pool

pkg/barback/lib/src/file_pool.dart

+// Copyright (c) 2013, 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 barback.file_pool;
+
+import 'dart:async';
+import 'dart:collection';
+import 'dart:io';
+
+/// Manages a pool of files that are opened for reading to cope with maximum
+/// file descriptor limits.
+///
+/// If a file cannot be opened because too many files are already open, this
+/// will defer the open until a previously opened file is closed and then try
+/// again. If this doesn't succeed after a certain amount of time, the open
+/// will fail and the original "too many files" exception will be thrown.
+class FilePool {
+  // TODO(rnystrom): Should we cap this to some maximum size?
+  final _pendingListens = new Queue<_FileReader>();
+
+  /// Opens [file] for reading.
+  ///
+  /// When the returned stream is listened to, if there are too many files
+  /// open, this will wait for a previously opened to file to be closed and
+  /// then try again.
+  Stream<List<int>> openRead(File file) => new _FileReader(this, file).stream;
+
+  Future<String> readAsString(File file, Encoding encoding) {
+    return _readAsBytes(file).then(encoding.decode);
+  }
+
+  Future<List<int>> _readAsBytes(File file) {
+    Completer<List<int>> completer = new Completer<List<int>>();
+    var builder = new BytesBuilder();
+    openRead(file).listen(
+      (d) => builder.add(d),
+      onDone: () {
+        completer.complete(builder.takeBytes());
+      },
+      onError: (e, StackTrace stackTrace) {
+        completer.completeError(e, stackTrace);
+      },
+      cancelOnError: true);
+    return completer.future;
+  }
+
+  void _retryPendingListen() {
+    if (_pendingListens.isEmpty) return;
+
+    var pending = _pendingListens.removeFirst();
+    pending._listen();
+  }
+}
+
+/// Wraps a raw file reading stream in a stream that handles "too many files"
+/// errors.
+///
+/// This also notifies the pool when the underlying file stream is closed so
+/// that it can try to open a waiting file.
+class _FileReader {
+  final FilePool _pool;
+  final File _file;
+
+  /// The underyling file stream.
+  Stream<List<int>> _fileStream;
+
+  /// The controller for the wrapped stream.
+  StreamController<List<int>> _controller;
+
+  /// The current subscription to the underlying file stream.
+  ///
+  /// This will only be non-null while the wrapped stream is being listened to.
+  StreamSubscription _subscription;
+  Timer _timer;
+
+  /// When a [listen] call has thrown a "too many files" error, this will be
+  /// the exception object.
+  Object _exception;
+
+  /// When a [listen] call has thrown a "too many files" error, this will be
+  /// the captured stack trace.
+  Object _stackTrace;
+
+  /// The wrapped stream that the file can be read from.
+  Stream<List<int>> get stream {
+    if (_controller != null) return _controller.stream;
+
+    _controller = new StreamController<List<int>>(onListen: _listen,
+        onPause: () {
+      _subscription.pause();
+    }, onResume: () {
+      _subscription.resume();
+    }, onCancel: () {
+      if (_subscription != null) _subscription.cancel();
+      _subscription = null;
+    }, sync: true);
+
+    return _controller.stream;
+  }
+
+  _FileReader(this._pool, this._file);
+
+  /// Starts listening to the underlying file stream.
+  void _listen() {
+    if (_timer != null) {
+      _timer.cancel();
+      _timer = null;
+    }
+
+    _fileStream = _file.openRead();
+    _subscription = _fileStream.listen(_controller.add,
+        onError: _onError, onDone: _onDone, cancelOnError: true);
+  }
+
+  /// Handles an error from the underlying file stream.
+  ///
+  /// "Too many file" errors are caught so that we can retry later. Other
+  /// errors are passed to the wrapped stream and the underlying stream
+  /// subscription is canceled.
+  void _onError(Object exception, Object stackTrace) {
+    assert(_subscription != null);
+    assert(_exception == null);
+
+    // The subscription is canceled after an error.
+    _subscription = null;
+
+    // We only handle "Too many open files errors".
+    if (exception is! FileException || exception.osError.errorCode != 24) {
+      // TODO(bob): stack trace.
+      _controller.addError(exception, stackTrace);
+      return;
+    }
+
+    // TODO(bob): What if already deferred?
+    _exception = exception;
+    _stackTrace = stackTrace;
+
+    // We'll try to defer the listen in the hopes that another file will close
+    // and we can try. If that doesn't happen after a while, give up and just
+    // throw the original error.
+    // TODO(bob): How long?
+    _timer = new Timer(new Duration(seconds: 5), _onTimeout);
+
+    // Tell the pool that this file is waiting.
+    _pool._pendingListens.add(this);
+  }
+
+  /// Handles the underlying file stream finishing.
+  void _onDone() {
+    _subscription = null;
+
+    _controller.close();
+    _pool._retryPendingListen();
+  }
+
+  /// If this file failed to be read because there were too many open files and
+  /// no file was closed in time to retry, this handles giving up.
+  void _onTimeout() {
+    assert(_subscription == null);
+    assert(_exception != null);
+
+    // We failed to open in time, so just fail with the original error.
+    _pool._pendingListens.remove(this);
+    _controller.addError(_exception, _stackTrace);
+    _controller.close();
+
+    _timer = null;
+    _exception = null;
+    _stackTrace = null;
+
+  }
+}