Format all dart: library files

sdk/lib/async/stream_controller.dart

 // Copyright (c) 2012, 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;
 
 // -------------------------------------------------------------------
 // Controller for creating and adding events to a stream.
 // -------------------------------------------------------------------
 
@@ skipping 73 matching lines @@
    *
    * The [onListen] callback is called when the stream
    * receives its listener and [onCancel] when the listener ends
    * its subscription. If [onCancel] needs to perform an asynchronous operation,
    * [onCancel] should return a future that completes when the cancel operation
    * is done.
    *
    * If the stream is canceled before the controller needs new data the
    * [onResume] call might not be executed.
    */
-  factory StreamController({void onListen(),
-                            void onPause(),
-                            void onResume(),
-                            onCancel(),
-                            bool sync: false}) {
+  factory StreamController(
+      {void onListen(),
+      void onPause(),
+      void onResume(),
+      onCancel(),
+      bool sync: false}) {
     return sync
-         ? new _SyncStreamController<T>(onListen, onPause, onResume, onCancel)
-         : new _AsyncStreamController<T>(onListen, onPause, onResume, onCancel);
+        ? new _SyncStreamController<T>(onListen, onPause, onResume, onCancel)
+        : new _AsyncStreamController<T>(onListen, onPause, onResume, onCancel);
   }
 
   /**
    * A controller where [stream] can be listened to more than once.
    *
    * The [Stream] returned by [stream] is a broadcast stream.
    * It can be listened to more than once.
    *
    * A Stream should be inert until a subscriber starts listening on it (using
    * the [onListen] callback to start producing events). Streams should not
@@ skipping 33 matching lines @@
    * A listener must be subscribed both when the event is initiated
    * (that is, when [add] is called)
    * and when the event is later delivered,
    * in order to receive the event.
    *
    * The [onListen] callback is called when the first listener is subscribed,
    * and the [onCancel] is called when there are no longer any active listeners.
    * If a listener is added again later, after the [onCancel] was called,
    * the [onListen] will be called again.
    */
-  factory StreamController.broadcast({void onListen(),
-                                      void onCancel(),
-                                      bool sync: false}) {
+  factory StreamController.broadcast(
+      {void onListen(), void onCancel(), bool sync: false}) {
     return sync
         ? new _SyncBroadcastStreamController<T>(onListen, onCancel)
         : new _AsyncBroadcastStreamController<T>(onListen, onCancel);
   }
 
   /**
    * The callback which is called when the stream is listened to.
    *
    * May be set to `null`, in which case no callback will happen.
    */
@@ skipping 87 matching lines @@
    * returned future.
    *
    * If [cancelOnError] is true, only the first error on [source] is
    * forwarded to the controller's stream, and the `addStream` ends
    * after this. If [cancelOnError] is false, all errors are forwarded
    * and only a done event will end the `addStream`.
    */
   Future addStream(Stream<T> source, {bool cancelOnError: true});
 }
 
-
 /**
  * A stream controller that delivers its events synchronously.
  *
  * A synchronous stream controller is intended for cases where
  * an already asynchronous event triggers an event on a stream.
  *
  * Instead of adding the event to the stream in a later microtask,
  * causing extra latency, the event is instead fired immediately by the
  * synchronous stream controller, as if the stream event was
  * the current event or microtask.
@@ skipping 79 matching lines @@
    * Closes the controller's stream.
    *
    * As [StreamController.close], but must not be called while an event is
    * being added by [add], [addError] or [close].
    */
   Future close();
 }
 
 abstract class _StreamControllerLifecycle<T> {
   StreamSubscription<T> _subscribe(
-      void onData(T data),
-      Function onError,
-      void onDone(),
-      bool cancelOnError);
+      void onData(T data), Function onError, void onDone(), bool cancelOnError);
   void _recordPause(StreamSubscription<T> subscription) {}
   void _recordResume(StreamSubscription<T> subscription) {}
   Future _recordCancel(StreamSubscription<T> subscription) => null;
 }
 
 /**
  * Default implementation of [StreamController].
  *
  * Controls a stream that only supports a single controller.
  */
-abstract class _StreamController<T> implements StreamController<T>,
-                                               _StreamControllerLifecycle<T>,
-                                               _EventSink<T>,
-                                               _EventDispatch<T> {
+abstract class _StreamController<T>
+    implements
+        StreamController<T>,
+        _StreamControllerLifecycle<T>,
+        _EventSink<T>,
+        _EventDispatch<T> {
   // The states are bit-flags. More than one can be set at a time.
   //
   // The "subscription state" goes through the states:
   //   initial -> subscribed -> canceled.
   // These are mutually exclusive.
   // The "closed" state records whether the [close] method has been called
   // on the controller. This can be done at any time. If done before
   // subscription, the done event is queued. If done after cancel, the done
   // event is ignored (just as any other event after a cancel).
 
@@ skipping 50 matching lines @@
   // TODO(lrn): Could this be stored in the varData field too, if it's not
   // accessed until the call to "close"? Then we need to special case if it's
   // accessed earlier, or if close is called before subscribing.
   _Future _doneFuture;
 
   ControllerCallback onListen;
   ControllerCallback onPause;
   ControllerCallback onResume;
   ControllerCancelCallback onCancel;
 
-  _StreamController(this.onListen,
-                    this.onPause,
-                    this.onResume,
-                    this.onCancel);
+  _StreamController(this.onListen, this.onPause, this.onResume, this.onCancel);
 
   // Return a new stream every time. The streams are equal, but not identical.
   Stream<T> get stream => new _ControllerStream<T>(this);
 
   /**
    * Returns a view of this object that only exposes the [StreamSink] interface.
    */
   StreamSink<T> get sink => new _StreamSinkWrapper<T>(this);
 
   /**
    * Whether a listener has existed and been canceled.
    *
    * After this, adding more events will be ignored.
    */
   bool get _isCanceled => (_state & _STATE_CANCELED) != 0;
 
   /** Whether there is an active listener. */
   bool get hasListener => (_state & _STATE_SUBSCRIBED) != 0;
 
   /** Whether there has not been a listener yet. */
   bool get _isInitialState =>
       (_state & _STATE_SUBSCRIPTION_MASK) == _STATE_INITIAL;
 
   bool get isClosed => (_state & _STATE_CLOSED) != 0;
 
-  bool get isPaused => hasListener ? _subscription._isInputPaused
-                                   : !_isCanceled;
+  bool get isPaused =>
+      hasListener ? _subscription._isInputPaused : !_isCanceled;
 
   bool get _isAddingStream => (_state & _STATE_ADDSTREAM) != 0;
 
   /** New events may not be added after close, or during addStream. */
   bool get _mayAddEvent => (_state < _STATE_CLOSED);
 
   // Returns the pending events.
   // Pending events are events added before a subscription exists.
   // They are added to the subscription when it is created.
   // Pending events, if any, are kept in the _varData field until the
   // stream is listened to.
   // While adding a stream, pending events are moved into the
   // state object to allow the state object to use the _varData field.
   _PendingEvents<T> get _pendingEvents {
     assert(_isInitialState);
     if (!_isAddingStream) {
-      return _varData as Object /*=_PendingEvents<T>*/;
+      return _varData as Object/*=_PendingEvents<T>*/;
     }
     _StreamControllerAddStreamState<T> state =
-      _varData as Object /*=_StreamControllerAddStreamState<T>*/;
-    return state.varData as Object /*=_PendingEvents<T>*/;
+        _varData as Object/*=_StreamControllerAddStreamState<T>*/;
+    return state.varData as Object/*=_PendingEvents<T>*/;
   }
 
   // Returns the pending events, and creates the object if necessary.
   _StreamImplEvents<T> _ensurePendingEvents() {
     assert(_isInitialState);
     if (!_isAddingStream) {
       if (_varData == null) _varData = new _StreamImplEvents<T>();
-      return _varData as Object /*=_StreamImplEvents<T>*/;
+      return _varData as Object/*=_StreamImplEvents<T>*/;
     }
     _StreamControllerAddStreamState<T> state =
-        _varData as Object /*=_StreamControllerAddStreamState<T>*/;
+        _varData as Object/*=_StreamControllerAddStreamState<T>*/;
     if (state.varData == null) state.varData = new _StreamImplEvents<T>();
-    return state.varData as Object /*=_StreamImplEvents<T>*/;
+    return state.varData as Object/*=_StreamImplEvents<T>*/;
   }
 
   // Get the current subscription.
   // If we are adding a stream, the subscription is moved into the state
   // object to allow the state object to use the _varData field.
   _ControllerSubscription<T> get _subscription {
     assert(hasListener);
     if (_isAddingStream) {
       _StreamControllerAddStreamState<T> addState =
-          _varData as Object /*=_StreamControllerAddStreamState<T>*/;
-      return addState.varData as Object /*=_ControllerSubscription<T>*/;
+          _varData as Object/*=_StreamControllerAddStreamState<T>*/;
+      return addState.varData as Object/*=_ControllerSubscription<T>*/;
     }
-    return _varData as Object /*=_ControllerSubscription<T>*/;
+    return _varData as Object/*=_ControllerSubscription<T>*/;
   }
 
   /**
    * Creates an error describing why an event cannot be added.
    *
    * The reason, and therefore the error message, depends on the current state.
    */
   Error _badEventState() {
     if (isClosed) {
       return new StateError("Cannot add event after closing");
     }
     assert(_isAddingStream);
     return new StateError("Cannot add event while adding a stream");
   }
 
   // StreamSink interface.
   Future addStream(Stream<T> source, {bool cancelOnError: true}) {
     if (!_mayAddEvent) throw _badEventState();
     if (_isCanceled) return new _Future.immediate(null);
     _StreamControllerAddStreamState<T> addState =
-        new _StreamControllerAddStreamState<T>(this,
-                                               _varData,
-                                               source,
-                                               cancelOnError);
+        new _StreamControllerAddStreamState<T>(
+            this, _varData, source, cancelOnError);
     _varData = addState;
     _state |= _STATE_ADDSTREAM;
     return addState.addStreamFuture;
   }
 
   /**
    * Returns a future that is completed when the stream is done
    * processing events.
    *
    * This happens either when the done event has been sent, or if the
@@ skipping 78 matching lines @@
       _sendError(error, stackTrace);
     } else if (_isInitialState) {
       _ensurePendingEvents().add(new _DelayedError(error, stackTrace));
     }
   }
 
   void _close() {
     // End of addStream stream.
     assert(_isAddingStream);
     _StreamControllerAddStreamState<T> addState =
-        _varData as Object /*=_StreamControllerAddStreamState<T>*/;
+        _varData as Object/*=_StreamControllerAddStreamState<T>*/;
     _varData = addState.varData;
     _state &= ~_STATE_ADDSTREAM;
     addState.complete();
   }
 
   // _StreamControllerLifeCycle interface
 
-  StreamSubscription<T> _subscribe(
-      void onData(T data),
-      Function onError,
-      void onDone(),
-      bool cancelOnError) {
+  StreamSubscription<T> _subscribe(void onData(T data), Function onError,
+      void onDone(), bool cancelOnError) {
     if (!_isInitialState) {
       throw new StateError("Stream has already been listened to.");
     }
-    _ControllerSubscription<T> subscription =
-        new _ControllerSubscription<T>(this, onData, onError, onDone,
-                                       cancelOnError);
+    _ControllerSubscription<T> subscription = new _ControllerSubscription<T>(
+        this, onData, onError, onDone, cancelOnError);
 
     _PendingEvents<T> pendingEvents = _pendingEvents;
     _state |= _STATE_SUBSCRIBED;
     if (_isAddingStream) {
       _StreamControllerAddStreamState<T> addState =
-          _varData as Object /*=_StreamControllerAddStreamState<T>*/;
+          _varData as Object/*=_StreamControllerAddStreamState<T>*/;
       addState.varData = subscription;
       addState.resume();
     } else {
       _varData = subscription;
     }
     subscription._setPendingEvents(pendingEvents);
     subscription._guardCallback(() {
       _runGuarded(onListen);
     });
 
     return subscription;
   }
 
   Future _recordCancel(StreamSubscription<T> subscription) {
     // When we cancel, we first cancel any stream being added,
     // Then we call `onCancel`, and finally the _doneFuture is completed.
     // If either of addStream's cancel or `onCancel` returns a future,
     // we wait for it before continuing.
     // Any error during this process ends up in the returned future.
     // If more errors happen, we act as if it happens inside nested try/finallys
     // or whenComplete calls, and only the last error ends up in the
     // returned future.
     Future result;
     if (_isAddingStream) {
       _StreamControllerAddStreamState<T> addState =
-          _varData as Object /*=_StreamControllerAddStreamState<T>*/;
+          _varData as Object/*=_StreamControllerAddStreamState<T>*/;
       result = addState.cancel();
     }
     _varData = null;
     _state =
         (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED;
 
     if (onCancel != null) {
       if (result == null) {
         // Only introduce a future if one is needed.
         // If _onCancel returns null, no future is needed.
@@ skipping 22 matching lines @@
     } else {
       complete();
     }
 
     return result;
   }
 
   void _recordPause(StreamSubscription<T> subscription) {
     if (_isAddingStream) {
       _StreamControllerAddStreamState<T> addState =
-          _varData as Object /*=_StreamControllerAddStreamState<T>*/;
+          _varData as Object/*=_StreamControllerAddStreamState<T>*/;
       addState.pause();
     }
     _runGuarded(onPause);
   }
 
   void _recordResume(StreamSubscription<T> subscription) {
     if (_isAddingStream) {
       _StreamControllerAddStreamState<T> addState =
-          _varData as Object /*=_StreamControllerAddStreamState<T>*/;
+          _varData as Object/*=_StreamControllerAddStreamState<T>*/;
       addState.resume();
     }
     _runGuarded(onResume);
   }
 }
 
 abstract class _SyncStreamControllerDispatch<T>
     implements _StreamController<T>, SynchronousStreamController<T> {
   int get _state;
   void set _state(int state);
@@ skipping 23 matching lines @@
 
   void _sendDone() {
     _subscription._addPending(const _DelayedDone());
   }
 }
 
 // TODO(lrn): Use common superclass for callback-controllers when VM supports
 // constructors in mixin superclasses.
 
 class _AsyncStreamController<T> = _StreamController<T>
-                                  with _AsyncStreamControllerDispatch<T>;
+    with _AsyncStreamControllerDispatch<T>;
 
 class _SyncStreamController<T> = _StreamController<T>
-                                 with _SyncStreamControllerDispatch<T>;
+    with _SyncStreamControllerDispatch<T>;
 
 typedef _NotificationHandler();
 
 Future _runGuarded(_NotificationHandler notificationHandler) {
   if (notificationHandler == null) return null;
   try {
     var result = notificationHandler();
     if (result is Future) return result;
     return null;
   } catch (e, s) {
     Zone.current.handleUncaughtError(e, s);
   }
 }
 
 class _ControllerStream<T> extends _StreamImpl<T> {
   _StreamControllerLifecycle<T> _controller;
 
   _ControllerStream(this._controller);
 
-  StreamSubscription<T> _createSubscription(
-      void onData(T data),
-      Function onError,
-      void onDone(),
-      bool cancelOnError) =>
-    _controller._subscribe(onData, onError, onDone, cancelOnError);
+  StreamSubscription<T> _createSubscription(void onData(T data),
+          Function onError, void onDone(), bool cancelOnError) =>
+      _controller._subscribe(onData, onError, onDone, cancelOnError);
 
   // Override == and hashCode so that new streams returned by the same
   // controller are considered equal. The controller returns a new stream
   // each time it's queried, but doesn't have to cache the result.
 
   int get hashCode => _controller.hashCode ^ 0x35323532;
 
-  bool operator==(Object other) {
+  bool operator ==(Object other) {
     if (identical(this, other)) return true;
     if (other is! _ControllerStream) return false;
     _ControllerStream otherStream = other;
     return identical(otherStream._controller, this._controller);
   }
 }
 
 class _ControllerSubscription<T> extends _BufferingStreamSubscription<T> {
   final _StreamControllerLifecycle<T> _controller;
 
   _ControllerSubscription(this._controller, void onData(T data),
-                          Function onError, void onDone(), bool cancelOnError)
+      Function onError, void onDone(), bool cancelOnError)
       : super(onData, onError, onDone, cancelOnError);
 
   Future _onCancel() {
     return _controller._recordCancel(this);
   }
 
   void _onPause() {
     _controller._recordPause(this);
   }
 
   void _onResume() {
     _controller._recordResume(this);
   }
 }
 
-
 /** A class that exposes only the [StreamSink] interface of an object. */
 class _StreamSinkWrapper<T> implements StreamSink<T> {
   final StreamController _target;
   _StreamSinkWrapper(this._target);
-  void add(T data) { _target.add(data); }
+  void add(T data) {
+    _target.add(data);
+  }
+
   void addError(Object error, [StackTrace stackTrace]) {
     _target.addError(error, stackTrace);
   }
+
   Future close() => _target.close();
   Future addStream(Stream<T> source, {bool cancelOnError: true}) =>
       _target.addStream(source, cancelOnError: cancelOnError);
   Future get done => _target.done;
 }
 
 /**
  * Object containing the state used to handle [StreamController.addStream].
  */
 class _AddStreamState<T> {
   // [_Future] returned by call to addStream.
   final _Future addStreamFuture;
 
   // Subscription on stream argument to addStream.
   final StreamSubscription addSubscription;
 
   _AddStreamState(_EventSink<T> controller, Stream source, bool cancelOnError)
       : addStreamFuture = new _Future(),
         addSubscription = source.listen(controller._add,
-                                        onError: cancelOnError
-                                             ? makeErrorHandler(controller)
-                                             : controller._addError,
-                                        onDone: controller._close,
-                                        cancelOnError: cancelOnError);
+            onError: cancelOnError
+                ? makeErrorHandler(controller)
+                : controller._addError,
+            onDone: controller._close,
+            cancelOnError: cancelOnError);
 
-  static makeErrorHandler(_EventSink controller) =>
-      (e, StackTrace s) {
+  static makeErrorHandler(_EventSink controller) => (e, StackTrace s) {
         controller._addError(e, s);
         controller._close();
       };
 
   void pause() {
     addSubscription.pause();
   }
 
   void resume() {
     addSubscription.resume();
   }
 
   /**
    * Stop adding the stream.
    *
    * Complete the future returned by `StreamController.addStream` when
    * the cancel is complete.
    *
    * Return a future if the cancel takes time, otherwise return `null`.
    */
   Future cancel() {
     var cancel = addSubscription.cancel();
     if (cancel == null) {
       addStreamFuture._asyncComplete(null);
       return null;
     }
-    return cancel.whenComplete(() { addStreamFuture._asyncComplete(null); });
+    return cancel.whenComplete(() {
+      addStreamFuture._asyncComplete(null);
+    });
   }
 
   void complete() {
     addStreamFuture._asyncComplete(null);
   }
 }
 
 class _StreamControllerAddStreamState<T> extends _AddStreamState<T> {
   // The subscription or pending data of a _StreamController.
   // Stored here because we reuse the `_varData` field  in the _StreamController
   // to store this state object.
   var varData;
 
-  _StreamControllerAddStreamState(_StreamController<T> controller,
-                                  this.varData,
-                                  Stream source,
-                                  bool cancelOnError)
+  _StreamControllerAddStreamState(_StreamController<T> controller, this.varData,
+      Stream source, bool cancelOnError)
       : super(controller, source, cancelOnError) {
     if (controller.isPaused) {
       addSubscription.pause();
     }
   }
 }