Format all dart: library files

sdk/lib/async/stream.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.
 
 part of dart.async;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 86 matching lines @@
    *
    * When the future completes, the stream will fire one event, either
    * data or error, and then close with a done-event.
    */
   factory Stream.fromFuture(Future<T> future) {
     // Use the controller's buffering to fill in the value even before
     // the stream has a listener. For a single value, it's not worth it
     // to wait for a listener before doing the `then` on the future.
     _StreamController<T> controller = new StreamController<T>(sync: true);
     future.then((value) {
-        controller._add(value);
-        controller._closeUnchecked();
-      },
-      onError: (error, stackTrace) {
-        controller._addError(error, stackTrace);
-        controller._closeUnchecked();
-      });
+      controller._add(value);
+      controller._closeUnchecked();
+    }, onError: (error, stackTrace) {
+      controller._addError(error, stackTrace);
+      controller._closeUnchecked();
+    });
     return controller.stream;
   }
 
   /**
    * Create a stream from a group of futures.
    *
    * The stream reports the results of the futures on the stream in the order
    * in which the futures complete.
    *
    * If some futures have completed before calling `Stream.fromFutures`,
@@ skipping 50 matching lines @@
   /**
    * Creates a stream that repeatedly emits events at [period] intervals.
    *
    * The event values are computed by invoking [computation]. The argument to
    * this callback is an integer that starts with 0 and is incremented for
    * every event.
    *
    * If [computation] is omitted the event values will all be `null`.
    */
   factory Stream.periodic(Duration period,
-                          [T computation(int computationCount)]) {
+      [T computation(int computationCount)]) {
     Timer timer;
     int computationCount = 0;
     StreamController<T> controller;
     // Counts the time that the Stream was running (and not paused).
     Stopwatch watch = new Stopwatch();
 
     void sendEvent() {
       watch.reset();
       T data;
       if (computation != null) {
         try {
           data = computation(computationCount++);
         } catch (e, s) {
           controller.addError(e, s);
           return;
         }
       }
       controller.add(data);
     }
 
     void startPeriodicTimer() {
       assert(timer == null);
       timer = new Timer.periodic(period, (Timer timer) {
         sendEvent();
       });
     }
 
-    controller = new StreamController<T>(sync: true,
+    controller = new StreamController<T>(
+        sync: true,
         onListen: () {
           watch.start();
           startPeriodicTimer();
         },
         onPause: () {
           timer.cancel();
           timer = null;
           watch.stop();
         },
         onResume: () {
@@ skipping 47 matching lines @@
    *       // Some generic types ommitted for brevety.
    *       Stream bind(Stream stream) => new Stream<String>.eventTransformed(
    *           stream,
    *           (EventSink sink) => new DuplicationSink(sink));
    *     }
    *
    *     stringStream.transform(new DuplicationTransformer());
    *
    * The resulting stream is a broadcast stream if [source] is.
    */
-  factory Stream.eventTransformed(Stream source,
-                                  EventSink mapSink(EventSink<T> sink)) {
+  factory Stream.eventTransformed(
+      Stream source, EventSink mapSink(EventSink<T> sink)) {
     return new _BoundSinkStream(source, mapSink);
   }
 
   /**
    * Reports whether this stream is a broadcast stream.
    */
   bool get isBroadcast => false;
 
   /**
    * Returns a multi-subscription stream that produces the same events as this.
    *
    * The returned stream will subscribe to this stream when its first
    * subscriber is added, and will stay subscribed until this stream ends,
    * or a callback cancels the subscription.
    *
    * If [onListen] is provided, it is called with a subscription-like object
    * that represents the underlying subscription to this stream. It is
    * possible to pause, resume or cancel the subscription during the call
    * to [onListen]. It is not possible to change the event handlers, including
    * using [StreamSubscription.asFuture].
    *
    * If [onCancel] is provided, it is called in a similar way to [onListen]
    * when the returned stream stops having listener. If it later gets
    * a new listener, the [onListen] function is called again.
    *
    * Use the callbacks, for example, for pausing the underlying subscription
    * while having no subscribers to prevent losing events, or canceling the
    * subscription when there are no listeners.
    */
-  Stream<T> asBroadcastStream({
-      void onListen(StreamSubscription<T> subscription),
-      void onCancel(StreamSubscription<T> subscription) }) {
+  Stream<T> asBroadcastStream(
+      {void onListen(StreamSubscription<T> subscription),
+      void onCancel(StreamSubscription<T> subscription)}) {
     return new _AsBroadcastStream<T>(this, onListen, onCancel);
   }
 
   /**
    * Adds a subscription to this stream.
    *
    * Returns a [StreamSubscription] which handles events from the stream using
    * the provided [onData], [onError] and [onDone] handlers.
    * The handlers can be changed on the subscription, but they start out
    * as the provided functions.
@@ skipping 19 matching lines @@
    * called. If [onDone] is `null`, nothing happens.
    *
    * If [cancelOnError] is true, the subscription is automatically cancelled
    * when the first error event is delivered. The default is `false`.
    *
    * While a subscription is paused, or when it has been cancelled,
    * the subscription doesn't receive events and none of the
    * event handler functions are called.
    */
   StreamSubscription<T> listen(void onData(T event),
-                               { Function onError,
-                                 void onDone(),
-                                 bool cancelOnError});
+      {Function onError, void onDone(), bool cancelOnError});
 
   /**
    * Creates a new stream from this stream that discards some data events.
    *
    * The new stream sends the same error and done events as this stream,
    * but it only sends the data events that satisfy the [test].
    *
    * The returned stream is a broadcast stream if this stream is.
    * If a broadcast stream is listened to more than once, each subscription
    * will individually perform the `test`.
@@ skipping 32 matching lines @@
    *
    * The returned stream is a broadcast stream if this stream is.
    */
   Stream<E> asyncMap<E>(convert(T event)) {
     StreamController<E> controller;
     StreamSubscription<T> subscription;
 
     void onListen() {
       final add = controller.add;
       assert(controller is _StreamController ||
-             controller is _BroadcastStreamController);
-      final _EventSink<E> eventSink =
-          controller as Object /*=_EventSink<E>*/;
+          controller is _BroadcastStreamController);
+      final _EventSink<E> eventSink = controller as Object/*=_EventSink<E>*/;
       final addError = eventSink._addError;
-      subscription = this.listen(
-          (T event) {
-            dynamic newValue;
-            try {
-              newValue = convert(event);
-            } catch (e, s) {
-              controller.addError(e, s);
-              return;
-            }
-            if (newValue is Future) {
-              subscription.pause();
-              newValue.then(add, onError: addError)
-                      .whenComplete(subscription.resume);
-            } else {
-              controller.add(newValue as Object/*=E*/);
-            }
-          },
-          onError: addError,
-          onDone: controller.close
-      );
+      subscription = this.listen((T event) {
+        dynamic newValue;
+        try {
+          newValue = convert(event);
+        } catch (e, s) {
+          controller.addError(e, s);
+          return;
+        }
+        if (newValue is Future) {
+          subscription.pause();
+          newValue
+              .then(add, onError: addError)
+              .whenComplete(subscription.resume);
+        } else {
+          controller.add(newValue as Object/*=E*/);
+        }
+      }, onError: addError, onDone: controller.close);
     }
 
     if (this.isBroadcast) {
       controller = new StreamController<E>.broadcast(
-        onListen: onListen,
-        onCancel: () { subscription.cancel(); },
-        sync: true
-      );
+          onListen: onListen,
+          onCancel: () {
+            subscription.cancel();
+          },
+          sync: true);
     } else {
       controller = new StreamController<E>(
-        onListen: onListen,
-        onPause: () { subscription.pause(); },
-        onResume: () { subscription.resume(); },
-        onCancel: () => subscription.cancel(),
-        sync: true
-      );
+          onListen: onListen,
+          onPause: () {
+            subscription.pause();
+          },
+          onResume: () {
+            subscription.resume();
+          },
+          onCancel: () => subscription.cancel(),
+          sync: true);
     }
     return controller.stream;
   }
 
   /**
    * Creates a new stream with the events of a stream per original event.
    *
    * This acts like [expand], except that [convert] returns a [Stream]
    * instead of an [Iterable].
    * The events of the returned stream becomes the events of the returned
    * stream, in the order they are produced.
    *
    * If [convert] returns `null`, no value is put on the output stream,
    * just as if it returned an empty stream.
    *
    * The returned stream is a broadcast stream if this stream is.
    */
   Stream<E> asyncExpand<E>(Stream<E> convert(T event)) {
     StreamController<E> controller;
     StreamSubscription<T> subscription;
     void onListen() {
       assert(controller is _StreamController ||
-             controller is _BroadcastStreamController);
-      final _EventSink<E> eventSink =
-          controller as Object /*=_EventSink<E>*/;
-      subscription = this.listen(
-          (T event) {
-            Stream<E> newStream;
-            try {
-              newStream = convert(event);
-            } catch (e, s) {
-              controller.addError(e, s);
-              return;
-            }
-            if (newStream != null) {
-              subscription.pause();
-              controller.addStream(newStream)
-                        .whenComplete(subscription.resume);
-            }
-          },
-          onError: eventSink._addError,  // Avoid Zone error replacement.
-          onDone: controller.close
-      );
+          controller is _BroadcastStreamController);
+      final _EventSink<E> eventSink = controller as Object/*=_EventSink<E>*/;
+      subscription = this.listen((T event) {
+        Stream<E> newStream;
+        try {
+          newStream = convert(event);
+        } catch (e, s) {
+          controller.addError(e, s);
+          return;
+        }
+        if (newStream != null) {
+          subscription.pause();
+          controller.addStream(newStream).whenComplete(subscription.resume);
+        }
+      },
+          onError: eventSink._addError, // Avoid Zone error replacement.
+          onDone: controller.close);
     }
+
     if (this.isBroadcast) {
       controller = new StreamController<E>.broadcast(
-        onListen: onListen,
-        onCancel: () { subscription.cancel(); },
-        sync: true
-      );
+          onListen: onListen,
+          onCancel: () {
+            subscription.cancel();
+          },
+          sync: true);
     } else {
       controller = new StreamController<E>(
-        onListen: onListen,
-        onPause: () { subscription.pause(); },
-        onResume: () { subscription.resume(); },
-        onCancel: () => subscription.cancel(),
-        sync: true
-      );
+          onListen: onListen,
+          onPause: () {
+            subscription.pause();
+          },
+          onResume: () {
+            subscription.resume();
+          },
+          onCancel: () => subscription.cancel(),
+          sync: true);
     }
     return controller.stream;
   }
 
   /**
    * Creates a wrapper Stream that intercepts some errors from this stream.
    *
    * If this stream sends an error that matches [test], then it is intercepted
    * by the [onError] function.
    *
@@ skipping 12 matching lines @@
    * or simply return to make the stream forget the error.
    *
    * If you need to transform an error into a data event, use the more generic
    * [Stream.transform] to handle the event by writing a data event to
    * the output sink.
    *
    * The returned stream is a broadcast stream if this stream is.
    * If a broadcast stream is listened to more than once, each subscription
    * will individually perform the `test` and handle the error.
    */
-  Stream<T> handleError(Function onError, { bool test(error) }) {
+  Stream<T> handleError(Function onError, {bool test(error)}) {
     return new _HandleErrorStream<T>(this, onError, test);
   }
 
   /**
    * Creates a new stream from this stream that converts each element
    * into zero or more events.
    *
    * Each incoming event is converted to an [Iterable] of new events,
    * and each of these new events are then sent by the returned stream
    * in order.
@@ skipping 29 matching lines @@
   }
 
   /**
    * Chains this stream as the input of the provided [StreamTransformer].
    *
    * Returns the result of [:streamTransformer.bind:] itself.
    *
    * The `streamTransformer` can decide whether it wants to return a
    * broadcast stream or not.
    */
-  Stream<S> transform<S>(
-      StreamTransformer<T, S > streamTransformer) {
+  Stream<S> transform<S>(StreamTransformer<T, S> streamTransformer) {
     return streamTransformer.bind(this);
   }
 
   /**
    * Reduces a sequence of values by repeatedly applying [combine].
    */
   Future<T> reduce(T combine(T previous, T element)) {
     _Future<T> result = new _Future<T>();
     bool seenFirst = false;
     T value;
     StreamSubscription subscription;
     subscription = this.listen(
-      (T element) {
-        if (seenFirst) {
-          _runUserCode(() => combine(value, element),
-                       (T newValue) { value = newValue; },
-                       _cancelAndErrorClosure(subscription, result));
-        } else {
-          value = element;
-          seenFirst = true;
-        }
-      },
-      onError: result._completeError,
-      onDone: () {
-        if (!seenFirst) {
-          try {
-            throw IterableElementError.noElement();
-          } catch (e, s) {
-            _completeWithErrorCallback(result, e,  s);
+        (T element) {
+          if (seenFirst) {
+            _runUserCode(() => combine(value, element), (T newValue) {
+              value = newValue;
+            }, _cancelAndErrorClosure(subscription, result));
+          } else {
+            value = element;
+            seenFirst = true;
           }
-        } else {
-          result._complete(value);
-        }
-      },
-      cancelOnError: true
-    );
+        },
+        onError: result._completeError,
+        onDone: () {
+          if (!seenFirst) {
+            try {
+              throw IterableElementError.noElement();
+            } catch (e, s) {
+              _completeWithErrorCallback(result, e, s);
+            }
+          } else {
+            result._complete(value);
+          }
+        },
+        cancelOnError: true);
     return result;
   }
 
   /** Reduces a sequence of values by repeatedly applying [combine]. */
-  Future<S> fold<S>(S initialValue,
-      S combine(S previous, T element)) {
-
+  Future<S> fold<S>(S initialValue, S combine(S previous, T element)) {
     _Future<S> result = new _Future<S>();
     S value = initialValue;
     StreamSubscription subscription;
-    subscription = this.listen(
-      (T element) {
-        _runUserCode(
-          () => combine(value, element),
-          (S newValue) { value = newValue; },
-          _cancelAndErrorClosure(subscription, result)
-        );
-      },
-      onError: (e, st) {
-        result._completeError(e, st);
-      },
-      onDone: () {
-        result._complete(value);
-      },
-      cancelOnError: true);
+    subscription = this.listen((T element) {
+      _runUserCode(() => combine(value, element), (S newValue) {
+        value = newValue;
+      }, _cancelAndErrorClosure(subscription, result));
+    }, onError: (e, st) {
+      result._completeError(e, st);
+    }, onDone: () {
+      result._complete(value);
+    }, cancelOnError: true);
     return result;
   }
 
   /**
    * Collects string of data events' string representations.
    *
    * If [separator] is provided, it is inserted between any two
    * elements.
    *
    * Any error in the stream causes the future to complete with that
    * error. Otherwise it completes with the collected string when
    * the "done" event arrives.
    */
   Future<String> join([String separator = ""]) {
     _Future<String> result = new _Future<String>();
     StringBuffer buffer = new StringBuffer();
     StreamSubscription subscription;
     bool first = true;
-    subscription = this.listen(
-      (T element) {
-        if (!first) {
-          buffer.write(separator);
-        }
-        first = false;
-        try {
-          buffer.write(element);
-        } catch (e, s) {
-          _cancelAndErrorWithReplacement(subscription, result, e, s);
-        }
-      },
-      onError: (e) {
-        result._completeError(e);
-      },
-      onDone: () {
-        result._complete(buffer.toString());
-      },
-      cancelOnError: true);
+    subscription = this.listen((T element) {
+      if (!first) {
+        buffer.write(separator);
+      }
+      first = false;
+      try {
+        buffer.write(element);
+      } catch (e, s) {
+        _cancelAndErrorWithReplacement(subscription, result, e, s);
+      }
+    }, onError: (e) {
+      result._completeError(e);
+    }, onDone: () {
+      result._complete(buffer.toString());
+    }, cancelOnError: true);
     return result;
   }
 
   /**
    * Checks whether [needle] occurs in the elements provided by this stream.
    *
    * Completes the [Future] when the answer is known.
    * If this stream reports an error, the [Future] will report that error.
    */
   Future<bool> contains(Object needle) {
     _Future<bool> future = new _Future<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
-          _runUserCode(
-            () => (element == needle),
-            (bool isMatch) {
-              if (isMatch) {
-                _cancelAndValue(subscription, future, true);
-              }
-            },
-            _cancelAndErrorClosure(subscription, future)
-          );
+          _runUserCode(() => (element == needle), (bool isMatch) {
+            if (isMatch) {
+              _cancelAndValue(subscription, future, true);
+            }
+          }, _cancelAndErrorClosure(subscription, future));
         },
         onError: future._completeError,
         onDone: () {
           future._complete(false);
         },
         cancelOnError: true);
     return future;
   }
 
   /**
    * Executes [action] on each data event of the stream.
    *
    * Completes the returned [Future] when all events of the stream
    * have been processed. Completes the future with an error if the
    * stream has an error event, or if [action] throws.
    */
   Future forEach(void action(T element)) {
     _Future future = new _Future();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
-          _runUserCode(
-            () => action(element),
-            (_) {},
-            _cancelAndErrorClosure(subscription, future)
-          );
+          _runUserCode(() => action(element), (_) {},
+              _cancelAndErrorClosure(subscription, future));
         },
         onError: future._completeError,
         onDone: () {
           future._complete(null);
         },
         cancelOnError: true);
     return future;
   }
 
   /**
    * Checks whether [test] accepts all elements provided by this stream.
    *
    * Completes the [Future] when the answer is known.
    * If this stream reports an error, the [Future] will report that error.
    */
   Future<bool> every(bool test(T element)) {
     _Future<bool> future = new _Future<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
-          _runUserCode(
-            () => test(element),
-            (bool isMatch) {
-              if (!isMatch) {
-                _cancelAndValue(subscription, future, false);
-              }
-            },
-            _cancelAndErrorClosure(subscription, future)
-          );
+          _runUserCode(() => test(element), (bool isMatch) {
+            if (!isMatch) {
+              _cancelAndValue(subscription, future, false);
+            }
+          }, _cancelAndErrorClosure(subscription, future));
         },
         onError: future._completeError,
         onDone: () {
           future._complete(true);
         },
         cancelOnError: true);
     return future;
   }
 
   /**
    * Checks whether [test] accepts any element provided by this stream.
    *
    * Completes the [Future] when the answer is known.
    *
    * If this stream reports an error, the [Future] reports that error.
    *
    * Stops listening to the stream after the first matching element has been
    * found.
    *
    * Internally the method cancels its subscription after this element. This
    * means that single-subscription (non-broadcast) streams are closed and
    * cannot be reused after a call to this method.
    */
   Future<bool> any(bool test(T element)) {
     _Future<bool> future = new _Future<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         (T element) {
-          _runUserCode(
-            () => test(element),
-            (bool isMatch) {
-              if (isMatch) {
-                _cancelAndValue(subscription, future, true);
-              }
-            },
-            _cancelAndErrorClosure(subscription, future)
-          );
+          _runUserCode(() => test(element), (bool isMatch) {
+            if (isMatch) {
+              _cancelAndValue(subscription, future, true);
+            }
+          }, _cancelAndErrorClosure(subscription, future));
         },
         onError: future._completeError,
         onDone: () {
           future._complete(false);
         },
         cancelOnError: true);
     return future;
   }
 
-
   /** Counts the elements in the stream. */
   Future<int> get length {
     _Future<int> future = new _Future<int>();
     int count = 0;
     this.listen(
-      (_) { count++; },
-      onError: future._completeError,
-      onDone: () {
-        future._complete(count);
-      },
-      cancelOnError: true);
+        (_) {
+          count++;
+        },
+        onError: future._completeError,
+        onDone: () {
+          future._complete(count);
+        },
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Reports whether this stream contains any elements.
    *
    * Stops listening to the stream after the first element has been received.
    *
    * Internally the method cancels its subscription after the first element.
    * This means that single-subscription (non-broadcast) streams are closed and
    * cannot be reused after a call to this getter.
    */
   Future<bool> get isEmpty {
     _Future<bool> future = new _Future<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
-      (_) {
-        _cancelAndValue(subscription, future, false);
-      },
-      onError: future._completeError,
-      onDone: () {
-        future._complete(true);
-      },
-      cancelOnError: true);
+        (_) {
+          _cancelAndValue(subscription, future, false);
+        },
+        onError: future._completeError,
+        onDone: () {
+          future._complete(true);
+        },
+        cancelOnError: true);
     return future;
   }
 
   /** Collects the data of this stream in a [List]. */
   Future<List<T>> toList() {
     List<T> result = <T>[];
     _Future<List<T>> future = new _Future<List<T>>();
     this.listen(
-      (T data) {
-        result.add(data);
-      },
-      onError: future._completeError,
-      onDone: () {
-        future._complete(result);
-      },
-      cancelOnError: true);
+        (T data) {
+          result.add(data);
+        },
+        onError: future._completeError,
+        onDone: () {
+          future._complete(result);
+        },
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Collects the data of this stream in a [Set].
    *
    * The returned set is the same type as returned by `new Set<T>()`.
    * If another type of set is needed, either use [forEach] to add each
    * element to the set, or use
    * `toList().then((list) => new SomeOtherSet.from(list))`
    * to create the set.
    */
   Future<Set<T>> toSet() {
     Set<T> result = new Set<T>();
     _Future<Set<T>> future = new _Future<Set<T>>();
     this.listen(
-      (T data) {
-        result.add(data);
-      },
-      onError: future._completeError,
-      onDone: () {
-        future._complete(result);
-      },
-      cancelOnError: true);
+        (T data) {
+          result.add(data);
+        },
+        onError: future._completeError,
+        onDone: () {
+          future._complete(result);
+        },
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Discards all data on the stream, but signals when it's done or an error
    * occurred.
    *
    * When subscribing using [drain], cancelOnError will be true. This means
    * that the future will complete with the first error on the stream and then
    * cancel the subscription.
    *
    * In case of a `done` event the future completes with the given
    * [futureValue].
    */
-  Future<E> drain<E>([E futureValue])
-      => listen(null, cancelOnError: true).asFuture<E>(futureValue);
+  Future<E> drain<E>([E futureValue]) => listen(null, cancelOnError: true)
+      .asFuture<E>(futureValue);
 
   /**
    * Provides at most the first [count] data events of this stream.
    *
    * Forwards all events of this stream to the returned stream
    * until [count] data events have been forwarded or this stream ends,
    * then ends the returned stream with a done event.
    *
    * If this stream produces fewer than [count] data events before it's done,
    * so will the returned stream.
@@ skipping 93 matching lines @@
    * If this stream is empty (a done event occurs before the first data event),
    * the resulting future completes with a [StateError].
    *
    * Except for the type of the error, this method is equivalent to
    * [:this.elementAt(0):].
    */
   Future<T> get first {
     _Future<T> future = new _Future<T>();
     StreamSubscription subscription;
     subscription = this.listen(
-      (T value) {
-        _cancelAndValue(subscription, future, value);
-      },
-      onError: future._completeError,
-      onDone: () {
-        try {
-          throw IterableElementError.noElement();
-        } catch (e, s) {
-          _completeWithErrorCallback(future, e, s);
-        }
-      },
-      cancelOnError: true);
+        (T value) {
+          _cancelAndValue(subscription, future, value);
+        },
+        onError: future._completeError,
+        onDone: () {
+          try {
+            throw IterableElementError.noElement();
+          } catch (e, s) {
+            _completeWithErrorCallback(future, e, s);
+          }
+        },
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Returns the last element of the stream.
    *
    * If an error event occurs before the first data event, the resulting future
    * is completed with that error.
    *
    * If this stream is empty (a done event occurs before the first data event),
    * the resulting future completes with a [StateError].
    */
   Future<T> get last {
     _Future<T> future = new _Future<T>();
     T result = null;
     bool foundResult = false;
     listen(
-      (T value) {
-        foundResult = true;
-        result = value;
-      },
-      onError: future._completeError,
-      onDone: () {
-        if (foundResult) {
-          future._complete(result);
-          return;
-        }
-        try {
-          throw IterableElementError.noElement();
-        } catch (e, s) {
-          _completeWithErrorCallback(future, e, s);
-        }
-      },
-      cancelOnError: true);
+        (T value) {
+          foundResult = true;
+          result = value;
+        },
+        onError: future._completeError,
+        onDone: () {
+          if (foundResult) {
+            future._complete(result);
+            return;
+          }
+          try {
+            throw IterableElementError.noElement();
+          } catch (e, s) {
+            _completeWithErrorCallback(future, e, s);
+          }
+        },
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Returns the single element.
    *
    * If an error event occurs before or after the first data event, the
    * resulting future is completed with that error.
    *
    * If [this] is empty or has more than one element throws a [StateError].
    */
   Future<T> get single {
     _Future<T> future = new _Future<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
-      (T value) {
-        if (foundResult) {
-          // This is the second element we get.
+        (T value) {
+          if (foundResult) {
+            // This is the second element we get.
+            try {
+              throw IterableElementError.tooMany();
+            } catch (e, s) {
+              _cancelAndErrorWithReplacement(subscription, future, e, s);
+            }
+            return;
+          }
+          foundResult = true;
+          result = value;
+        },
+        onError: future._completeError,
+        onDone: () {
+          if (foundResult) {
+            future._complete(result);
+            return;
+          }
           try {
-            throw IterableElementError.tooMany();
+            throw IterableElementError.noElement();
           } catch (e, s) {
-            _cancelAndErrorWithReplacement(subscription, future, e, s);
+            _completeWithErrorCallback(future, e, s);
           }
-          return;
-        }
-        foundResult = true;
-        result = value;
-      },
-      onError: future._completeError,
-      onDone: () {
-        if (foundResult) {
-          future._complete(result);
-          return;
-        }
-        try {
-          throw IterableElementError.noElement();
-        } catch (e, s) {
-          _completeWithErrorCallback(future, e, s);
-        }
-      },
-      cancelOnError: true);
+        },
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the first element of this stream matching [test].
    *
    * Returns a future that is filled with the first element of this stream
    * that [test] returns true for.
    *
    * If no such element is found before this stream is done, and a
    * [defaultValue] function is provided, the result of calling [defaultValue]
    * becomes the value of the future.
    *
    * Stops listening to the stream after the first matching element has been
    * received.
    *
    * Internally the method cancels its subscription after the first element that
    * matches the predicate. This means that single-subscription (non-broadcast)
    * streams are closed and cannot be reused after a call to this method.
    *
    * If an error occurs, or if this stream ends without finding a match and
    * with no [defaultValue] function provided, the future will receive an
    * error.
    */
   Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) {
     _Future<dynamic> future = new _Future();
     StreamSubscription subscription;
     subscription = this.listen(
-      (T value) {
-        _runUserCode(
-          () => test(value),
-          (bool isMatch) {
+        (T value) {
+          _runUserCode(() => test(value), (bool isMatch) {
             if (isMatch) {
               _cancelAndValue(subscription, future, value);
             }
-          },
-          _cancelAndErrorClosure(subscription, future)
-        );
-      },
-      onError: future._completeError,
-      onDone: () {
-        if (defaultValue != null) {
-          _runUserCode(defaultValue, future._complete, future._completeError);
-          return;
-        }
-        try {
-          throw IterableElementError.noElement();
-        } catch (e, s) {
-          _completeWithErrorCallback(future, e, s);
-        }
-      },
-      cancelOnError: true);
+          }, _cancelAndErrorClosure(subscription, future));
+        },
+        onError: future._completeError,
+        onDone: () {
+          if (defaultValue != null) {
+            _runUserCode(defaultValue, future._complete, future._completeError);
+            return;
+          }
+          try {
+            throw IterableElementError.noElement();
+          } catch (e, s) {
+            _completeWithErrorCallback(future, e, s);
+          }
+        },
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the last element in this stream matching [test].
    *
    * As [firstWhere], except that the last matching element is found.
    * That means that the result cannot be provided before this stream
    * is done.
    */
   Future<dynamic> lastWhere(bool test(T element), {Object defaultValue()}) {
     _Future<dynamic> future = new _Future();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
-      (T value) {
-        _runUserCode(
-          () => true == test(value),
-          (bool isMatch) {
+        (T value) {
+          _runUserCode(() => true == test(value), (bool isMatch) {
             if (isMatch) {
               foundResult = true;
               result = value;
             }
-          },
-          _cancelAndErrorClosure(subscription, future)
-        );
-      },
-      onError: future._completeError,
-      onDone: () {
-        if (foundResult) {
-          future._complete(result);
-          return;
-        }
-        if (defaultValue != null) {
-          _runUserCode(defaultValue, future._complete, future._completeError);
-          return;
-        }
-        try {
-          throw IterableElementError.noElement();
-        } catch (e, s) {
-          _completeWithErrorCallback(future, e, s);
-        }
-      },
-      cancelOnError: true);
+          }, _cancelAndErrorClosure(subscription, future));
+        },
+        onError: future._completeError,
+        onDone: () {
+          if (foundResult) {
+            future._complete(result);
+            return;
+          }
+          if (defaultValue != null) {
+            _runUserCode(defaultValue, future._complete, future._completeError);
+            return;
+          }
+          try {
+            throw IterableElementError.noElement();
+          } catch (e, s) {
+            _completeWithErrorCallback(future, e, s);
+          }
+        },
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the single element in this stream matching [test].
    *
    * Like [lastMatch], except that it is an error if more than one
    * matching element occurs in the stream.
    */
   Future<T> singleWhere(bool test(T element)) {
     _Future<T> future = new _Future<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
-      (T value) {
-        _runUserCode(
-          () => true == test(value),
-          (bool isMatch) {
+        (T value) {
+          _runUserCode(() => true == test(value), (bool isMatch) {
             if (isMatch) {
               if (foundResult) {
                 try {
                   throw IterableElementError.tooMany();
                 } catch (e, s) {
                   _cancelAndErrorWithReplacement(subscription, future, e, s);
                 }
                 return;
               }
               foundResult = true;
               result = value;
             }
-          },
-          _cancelAndErrorClosure(subscription, future)
-        );
-      },
-      onError: future._completeError,
-      onDone: () {
-        if (foundResult) {
-          future._complete(result);
-          return;
-        }
-        try {
-          throw IterableElementError.noElement();
-        } catch (e, s) {
-          _completeWithErrorCallback(future, e, s);
-        }
-      },
-      cancelOnError: true);
+          }, _cancelAndErrorClosure(subscription, future));
+        },
+        onError: future._completeError,
+        onDone: () {
+          if (foundResult) {
+            future._complete(result);
+            return;
+          }
+          try {
+            throw IterableElementError.noElement();
+          } catch (e, s) {
+            _completeWithErrorCallback(future, e, s);
+          }
+        },
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Returns the value of the [index]th data event of this stream.
    *
    * Stops listening to the stream after the [index]th data event has been
    * received.
    *
    * Internally the method cancels its subscription after these elements. This
    * means that single-subscription (non-broadcast) streams are closed and
    * cannot be reused after a call to this method.
    *
    * If an error event occurs before the value is found, the future completes
    * with this error.
    *
    * If a done event occurs before the value is found, the future completes
    * with a [RangeError].
    */
   Future<T> elementAt(int index) {
     if (index is! int || index < 0) throw new ArgumentError(index);
     _Future<T> future = new _Future<T>();
     StreamSubscription subscription;
     int elementIndex = 0;
     subscription = this.listen(
-      (T value) {
-        if (index == elementIndex) {
-          _cancelAndValue(subscription, future, value);
-          return;
-        }
-        elementIndex += 1;
-      },
-      onError: future._completeError,
-      onDone: () {
-        future._completeError(
-            new RangeError.index(index, this, "index", null, elementIndex));
-      },
-      cancelOnError: true);
+        (T value) {
+          if (index == elementIndex) {
+            _cancelAndValue(subscription, future, value);
+            return;
+          }
+          elementIndex += 1;
+        },
+        onError: future._completeError,
+        onDone: () {
+          future._completeError(
+              new RangeError.index(index, this, "index", null, elementIndex));
+        },
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Creates a new stream with the same events as this stream.
    *
    * Whenever more than [timeLimit] passes between two events from this stream,
    * the [onTimeout] function is called.
    *
    * The countdown doesn't start until the returned stream is listened to.
@@ skipping 18 matching lines @@
     StreamSubscription<T> subscription;
     Timer timer;
     Zone zone;
     _TimerCallback timeout;
 
     void onData(T event) {
       timer.cancel();
       controller.add(event);
       timer = zone.createTimer(timeLimit, timeout);
     }
+
     void onError(error, StackTrace stackTrace) {
       timer.cancel();
       assert(controller is _StreamController ||
-             controller is _BroadcastStreamController);
+          controller is _BroadcastStreamController);
       dynamic eventSink = controller;
-      eventSink._addError(error, stackTrace);  // Avoid Zone error replacement.
+      eventSink._addError(error, stackTrace); // Avoid Zone error replacement.
       timer = zone.createTimer(timeLimit, timeout);
     }
+
     void onDone() {
       timer.cancel();
       controller.close();
     }
+
     void onListen() {
       // This is the onListen callback for of controller.
       // It runs in the same zone that the subscription was created in.
       // Use that zone for creating timers and running the onTimeout
       // callback.
       zone = Zone.current;
       if (onTimeout == null) {
         timeout = () {
-          controller.addError(new TimeoutException("No stream event",
-                                                   timeLimit), null);
+          controller.addError(
+              new TimeoutException("No stream event", timeLimit), null);
         };
       } else {
         // TODO(floitsch): the return type should be 'void', and the type
         // should be inferred.
         var registeredOnTimeout =
             zone.registerUnaryCallback<dynamic, EventSink<T>>(onTimeout);
         _ControllerEventSinkWrapper wrapper =
             new _ControllerEventSinkWrapper(null);
         timeout = () {
-          wrapper._sink = controller;  // Only valid during call.
+          wrapper._sink = controller; // Only valid during call.
           zone.runUnaryGuarded(registeredOnTimeout, wrapper);
           wrapper._sink = null;
         };
       }
 
       subscription = this.listen(onData, onError: onError, onDone: onDone);
       timer = zone.createTimer(timeLimit, timeout);
     }
+
     Future onCancel() {
       timer.cancel();
       Future result = subscription.cancel();
       subscription = null;
       return result;
     }
+
     controller = isBroadcast
         ? new _SyncBroadcastStreamController<T>(onListen, onCancel)
-        : new _SyncStreamController<T>(
-              onListen,
-              () {
-                // Don't null the timer, onCancel may call cancel again.
-                timer.cancel();
-                subscription.pause();
-              },
-              () {
-                subscription.resume();
-                timer = zone.createTimer(timeLimit, timeout);
-              },
-              onCancel);
+        : new _SyncStreamController<T>(onListen, () {
+            // Don't null the timer, onCancel may call cancel again.
+            timer.cancel();
+            subscription.pause();
+          }, () {
+            subscription.resume();
+            timer = zone.createTimer(timeLimit, timeout);
+          }, onCancel);
     return controller.stream;
   }
 }
 
 /**
  * A subscription on events from a [Stream].
  *
  * When you listen on a [Stream] using [Stream.listen],
  * a [StreamSubscription] object is returned.
  *
@@ skipping 97 matching lines @@
    *
    * In case of an error the subscription will automatically cancel (even
    * when it was listening with `cancelOnError` set to `false`).
    *
    * In case of a `done` event the future completes with the given
    * [futureValue].
    */
   Future<E> asFuture<E>([E futureValue]);
 }
 
-
 /**
  * An interface that abstracts creation or handling of [Stream] events.
  */
 abstract class EventSink<T> implements Sink<T> {
   /** Send a data event to a stream. */
   void add(T event);
 
   /** Send an async error to a stream. */
   void addError(errorEvent, [StackTrace stackTrace]);
 
   /** Close the sink. No further events can be added after closing. */
   void close();
 }
 
-
 /** [Stream] wrapper that only exposes the [Stream] interface. */
 class StreamView<T> extends Stream<T> {
   final Stream<T> _stream;
 
-  const StreamView(Stream<T> stream) : _stream = stream, super._internal();
+  const StreamView(Stream<T> stream)
+      : _stream = stream,
+        super._internal();
 
   bool get isBroadcast => _stream.isBroadcast;
 
   Stream<T> asBroadcastStream(
-      {void onListen(StreamSubscription<T> subscription),
-       void onCancel(StreamSubscription<T> subscription)})
-      => _stream.asBroadcastStream(onListen: onListen, onCancel: onCancel);
+          {void onListen(StreamSubscription<T> subscription),
+          void onCancel(StreamSubscription<T> subscription)}) =>
+      _stream.asBroadcastStream(onListen: onListen, onCancel: onCancel);
 
   StreamSubscription<T> listen(void onData(T value),
-                               { Function onError,
-                                 void onDone(),
-                                 bool cancelOnError }) {
-    return _stream.listen(onData, onError: onError, onDone: onDone,
-                          cancelOnError: cancelOnError);
+      {Function onError, void onDone(), bool cancelOnError}) {
+    return _stream.listen(onData,
+        onError: onError, onDone: onDone, cancelOnError: cancelOnError);
   }
 }
 
-
 /**
  * Abstract interface for a "sink" accepting multiple entire streams.
  *
  * A consumer can accept a number of consecutive streams using [addStream],
  * and when no further data need to be added, the [close] method tells the
  * consumer to complete its work and shut down.
  *
  * This class is not just a [Sink<Stream>] because it is also combined with
  * other [Sink] classes, like it's combined with [EventSink] in the
  * [StreamSink] class.
@@ skipping 29 matching lines @@
    * This allows the consumer to complete any remaining work and release
    * resources that are no longer needed
    *
    * Returns a future which is completed when the consumer has shut down.
    * If cleaning up can fail, the error may be reported in the returned future,
    * otherwise it completes with `null`.
    */
   Future close();
 }
 
-
 /**
  * A object that accepts stream events both synchronously and asynchronously.
  *
  * A [StreamSink] unifies the asynchronous methods from [StreamConsumer] and
  * the synchronous methods from [EventSink].
  *
  * The [EventSink] methods can't be used while the [addStream] is called.
  * As soon as the [addStream]'s [Future] completes with a value, the
  * [EventSink] methods can be used again.
  *
@@ skipping 38 matching lines @@
    *
    * Otherwise, the returned future will complete when either:
    *
    * * all events have been processed and the sink has been closed, or
    * * the sink has otherwise been stopped from handling more events
    *   (for example by cancelling a stream subscription).
    */
   Future get done;
 }
 
-
 /**
  * The target of a [Stream.transform] call.
  *
  * The [Stream.transform] call will pass itself to this object and then return
  * the resulting stream.
  *
  * It is good practice to write transformers that can be used multiple times.
  */
 abstract class StreamTransformer<S, T> {
   /**
@@ skipping 53 matching lines @@
    *                 cancelOnError: cancelOnError);
    *             },
    *             onPause: () { subscription.pause(); },
    *             onResume: () { subscription.resume(); },
    *             onCancel: () => subscription.cancel(),
    *             sync: true);
    *           return controller.stream.listen(null);
    *         });
    */
   const factory StreamTransformer(
-      StreamSubscription<T> transformer(Stream<S> stream, bool cancelOnError))
-      = _StreamSubscriptionTransformer<S, T>;
+          StreamSubscription<T> transformer(
+              Stream<S> stream, bool cancelOnError)) =
+      _StreamSubscriptionTransformer<S, T>;
 
   /**
    * Creates a [StreamTransformer] that delegates events to the given functions.
    *
    * Example use of a duplicating transformer:
    *
    *     stringStream.transform(new StreamTransformer<String, String>.fromHandlers(
    *         handleData: (String value, EventSink<String> sink) {
    *           sink.add(value);
    *           sink.add(value);  // Duplicate the incoming events.
    *         }));
    */
-  factory StreamTransformer.fromHandlers({
-      void handleData(S data, EventSink<T> sink),
+  factory StreamTransformer.fromHandlers(
+      {void handleData(S data, EventSink<T> sink),
       void handleError(Object error, StackTrace stackTrace, EventSink<T> sink),
-      void handleDone(EventSink<T> sink)})
-          = _StreamHandlerTransformer<S, T>;
+      void handleDone(EventSink<T> sink)}) = _StreamHandlerTransformer<S, T>;
 
   /**
    * Transform the incoming [stream]'s events.
    *
    * Creates a new stream.
    * When this stream is listened to, it will start listening on [stream],
    * and generate events on the new stream based on the events from [stream].
    *
    * Subscriptions on the returned stream should propagate pause state
    * to the subscription on [stream].
    */
   Stream<T> bind(Stream<S> stream);
 }
 
 /**
  * An [Iterator] like interface for the values of a [Stream].
  *
  * This wraps a [Stream] and a subscription on the stream. It listens
  * on the stream, and completes the future returned by [moveNext] when the
  * next value becomes available.
  *
  * The stream may be paused between calls to [moveNext].
  */
 abstract class StreamIterator<T> {
-
   /** Create a [StreamIterator] on [stream]. */
   factory StreamIterator(Stream<T> stream)
       // TODO(lrn): use redirecting factory constructor when type
       // arguments are supported.
-      => new _StreamIterator<T>(stream);
+      =>
+      new _StreamIterator<T>(stream);
 
   /**
    * Wait for the next stream value to be available.
    *
    * Returns a future which will complete with either `true` or `false`.
    * Completing with `true` means that another event has been received and
    * can be read as [current].
    * Completing with `false` means that the stream iteration is done and
    * no further events will ever be available.
    * The future may complete with an error, if the stream produces an error,
@@ skipping 31 matching lines @@
    * If [moveNext] has been called when the iterator is canceled,
    * its returned future will complete with `false` as value,
    * as will all further calls to [moveNext].
    *
    * Returns a future if the cancel-operation is not completed synchronously.
    * Otherwise returns `null`.
    */
   Future cancel();
 }
 
-
 /**
  * Wraps an [_EventSink] so it exposes only the [EventSink] interface.
  */
 class _ControllerEventSinkWrapper<T> implements EventSink<T> {
   EventSink _sink;
   _ControllerEventSinkWrapper(this._sink);
 
-  void add(T data) { _sink.add(data); }
+  void add(T data) {
+    _sink.add(data);
+  }
+
   void addError(error, [StackTrace stackTrace]) {
     _sink.addError(error, stackTrace);
   }
-  void close() { _sink.close(); }
+
+  void close() {
+    _sink.close();
+  }
 }