Be sure we don't forward an error to a closed event stream.

R=rnystrom@google.com
BUG=15034

Review URL: https://codereview.chromium.org//68253010

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart/pkg/watcher@30210 260f80e4-7a28-3924-810f-c04153c831b5

pkgs/watcher/lib/src/async_queue.dart

diff --git a/pkgs/watcher/lib/src/async_queue.dart b/pkgs/watcher/lib/src/async_queue.dart
index 9456631..8ac0cdf 100644
--- a/pkgs/watcher/lib/src/async_queue.dart
+++ b/pkgs/watcher/lib/src/async_queue.dart
@@ -8,7 +8,6 @@
 import 'dart:collection';
 
 typedef Future ItemProcessor<T>(T item);
-typedef void ErrorHandler(error);
 
 /// A queue of items that are sequentially, asynchronously processed.
 ///
@@ -35,9 +34,9 @@
   /// The handler for errors thrown during processing.
   ///
   /// Used to avoid top-leveling asynchronous errors.
-  final ErrorHandler _errorHandler;
+  final Function _errorHandler;
 
-  AsyncQueue(this._processor, {ErrorHandler onError})
+  AsyncQueue(this._processor, {Function onError})
       : _errorHandler = onError;
 
   /// Enqueues [item] to be processed and starts asynchronously processing it

pkgs/watcher/lib/src/directory_watcher/polling.dart

diff --git a/pkgs/watcher/lib/src/directory_watcher/polling.dart b/pkgs/watcher/lib/src/directory_watcher/polling.dart
index 0e190b0..e50a0c0 100644
--- a/pkgs/watcher/lib/src/directory_watcher/polling.dart
+++ b/pkgs/watcher/lib/src/directory_watcher/polling.dart
@@ -72,7 +72,9 @@
 
   _PollingDirectoryWatcher(this.directory, this._pollingDelay) {
     _filesToProcess = new AsyncQueue<String>(_processFile,
-        onError: _events.addError);
+        onError: (e, stackTrace) {
+      if (!_events.isClosed) _events.addError(e, stackTrace);
+    });
 
     _poll();
   }