Format all dart: library files

sdk/lib/async/broadcast_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;
 
 class _BroadcastStream<T> extends _ControllerStream<T> {
   _BroadcastStream(_StreamControllerLifecycle<T> controller)
       : super(controller);
 
   bool get isBroadcast => true;
 }
 
 class _BroadcastSubscription<T> extends _ControllerSubscription<T> {
   static const int _STATE_EVENT_ID = 1;
   static const int _STATE_FIRING = 2;
   static const int _STATE_REMOVE_AFTER_FIRING = 4;
   // TODO(lrn): Use the _state field on _ControllerSubscription to
   // also store this state. Requires that the subscription implementation
   // does not assume that it's use of the state integer is the only use.
-  int _eventState = 0;  // Initialized to help dart2js type inference.
+  int _eventState = 0; // Initialized to help dart2js type inference.
 
   _BroadcastSubscription<T> _next;
   _BroadcastSubscription<T> _previous;
 
   _BroadcastSubscription(_StreamControllerLifecycle<T> controller,
-                         void onData(T data),
-                         Function onError,
-                         void onDone(),
-                         bool cancelOnError)
+      void onData(T data), Function onError, void onDone(), bool cancelOnError)
       : super(controller, onData, onError, onDone, cancelOnError) {
     _next = _previous = this;
   }
 
-  bool _expectsEvent(int eventId) =>
-      (_eventState & _STATE_EVENT_ID) == eventId;
+  bool _expectsEvent(int eventId) => (_eventState & _STATE_EVENT_ID) == eventId;
 
   void _toggleEventId() {
     _eventState ^= _STATE_EVENT_ID;
   }
 
   bool get _isFiring => (_eventState & _STATE_FIRING) != 0;
 
   void _setRemoveAfterFiring() {
     assert(_isFiring);
     _eventState |= _STATE_REMOVE_AFTER_FIRING;
   }
 
   bool get _removeAfterFiring =>
       (_eventState & _STATE_REMOVE_AFTER_FIRING) != 0;
 
   // The controller._recordPause doesn't do anything for a broadcast controller,
   // so we don't bother calling it.
-  void _onPause() { }
+  void _onPause() {}
 
   // The controller._recordResume doesn't do anything for a broadcast
   // controller, so we don't bother calling it.
-  void _onResume() { }
+  void _onResume() {}
 
   // _onCancel is inherited.
 }
 
 abstract class _BroadcastStreamController<T>
-    implements StreamController<T>,
-               _StreamControllerLifecycle<T>,
-               _EventSink<T>,
-               _EventDispatch<T> {
+    implements
+        StreamController<T>,
+        _StreamControllerLifecycle<T>,
+        _EventSink<T>,
+        _EventDispatch<T> {
   static const int _STATE_INITIAL = 0;
   static const int _STATE_EVENT_ID = 1;
   static const int _STATE_FIRING = 2;
   static const int _STATE_CLOSED = 4;
   static const int _STATE_ADDSTREAM = 8;
 
   ControllerCallback onListen;
   ControllerCancelCallback onCancel;
 
   // State of the controller.
@@ skipping 32 matching lines @@
   void set onPause(void onPauseHandler()) {
     throw new UnsupportedError(
         "Broadcast stream controllers do not support pause callbacks");
   }
 
   ControllerCallback get onResume {
     throw new UnsupportedError(
         "Broadcast stream controllers do not support pause callbacks");
   }
 
-  void set onResume(void onResumeHandler())  {
+  void set onResume(void onResumeHandler()) {
     throw new UnsupportedError(
         "Broadcast stream controllers do not support pause callbacks");
   }
 
   // StreamController interface.
 
   Stream<T> get stream => new _BroadcastStream<T>(this);
 
   StreamSink<T> get sink => new _StreamSinkWrapper<T>(this);
 
@@ skipping 68 matching lines @@
       _lastSubscription = previous;
     } else {
       next._previous = previous;
     }
 
     subscription._next = subscription._previous = subscription;
   }
 
   // _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 (isClosed) {
       if (onDone == null) onDone = _nullDoneHandler;
       return new _DoneStreamSubscription<T>(onDone);
     }
-    StreamSubscription<T> subscription =
-        new _BroadcastSubscription<T>(this, onData, onError, onDone,
-                                      cancelOnError);
+    StreamSubscription<T> subscription = new _BroadcastSubscription<T>(
+        this, onData, onError, onDone, cancelOnError);
     _addListener(subscription);
     if (identical(_firstSubscription, _lastSubscription)) {
       // Only one listener, so it must be the first listener.
       _runGuarded(onListen);
     }
     return subscription;
   }
 
   Future _recordCancel(StreamSubscription<T> sub) {
     _BroadcastSubscription<T> subscription = sub;
@@ skipping 124 matching lines @@
     assert(_isEmpty);
     if (isClosed && _doneFuture._mayComplete) {
       // When closed, _doneFuture is not null.
       _doneFuture._asyncComplete(null);
     }
     _runGuarded(onCancel);
   }
 }
 
 class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T>
-                                     implements SynchronousStreamController<T> {
+    implements SynchronousStreamController<T> {
   _SyncBroadcastStreamController(void onListen(), void onCancel())
       : super(onListen, onCancel);
 
   // EventDispatch interface.
 
   bool get _mayAddEvent => super._mayAddEvent && !_isFiring;
 
   _addEventError() {
     if (_isFiring) {
       return new StateError(
@@ skipping 40 matching lines @@
 }
 
 class _AsyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
   _AsyncBroadcastStreamController(void onListen(), void onCancel())
       : super(onListen, onCancel);
 
   // EventDispatch interface.
 
   void _sendData(T data) {
     for (_BroadcastSubscription<T> subscription = _firstSubscription;
-         subscription != null;
-         subscription = subscription._next) {
+        subscription != null;
+        subscription = subscription._next) {
       subscription._addPending(new _DelayedData<T>(data));
     }
   }
 
   void _sendError(Object error, StackTrace stackTrace) {
     for (_BroadcastSubscription<T> subscription = _firstSubscription;
-         subscription != null;
-         subscription = subscription._next) {
+        subscription != null;
+        subscription = subscription._next) {
       subscription._addPending(new _DelayedError(error, stackTrace));
     }
   }
 
   void _sendDone() {
     if (!_isEmpty) {
       for (_BroadcastSubscription<T> subscription = _firstSubscription;
-           subscription != null;
-           subscription = subscription._next) {
+          subscription != null;
+          subscription = subscription._next) {
         subscription._addPending(const _DelayedDone());
       }
     } else {
       assert(_doneFuture != null);
       assert(_doneFuture._mayComplete);
       _doneFuture._asyncComplete(null);
     }
   }
 }
 
 /**
  * Stream controller that is used by [Stream.asBroadcastStream].
  *
  * This stream controller allows incoming events while it is firing
  * other events. This is handled by delaying the events until the
  * current event is done firing, and then fire the pending events.
  *
  * This class extends [_SyncBroadcastStreamController]. Events of
  * an "asBroadcastStream" stream are always initiated by events
  * on another stream, and it is fine to forward them synchronously.
  */
-class _AsBroadcastStreamController<T>
-    extends _SyncBroadcastStreamController<T>
+class _AsBroadcastStreamController<T> extends _SyncBroadcastStreamController<T>
     implements _EventDispatch<T> {
   _StreamImplEvents<T> _pending;
 
   _AsBroadcastStreamController(void onListen(), void onCancel())
       : super(onListen, onCancel);
 
-  bool get _hasPending => _pending != null && ! _pending.isEmpty;
+  bool get _hasPending => _pending != null && !_pending.isEmpty;
 
   void _addPendingEvent(_DelayedEvent event) {
     if (_pending == null) {
       _pending = new _StreamImplEvents<T>();
     }
     _pending.add(event);
   }
 
   void add(T data) {
     if (!isClosed && _isFiring) {
@@ skipping 42 matching lines @@
 // It can simulate pauses, but otherwise does nothing.
 class _DoneSubscription<T> implements StreamSubscription<T> {
   int _pauseCount = 0;
   void onData(void handleData(T data)) {}
   void onError(Function handleError) {}
   void onDone(void handleDone()) {}
   void pause([Future resumeSignal]) {
     if (resumeSignal != null) resumeSignal.then(_resume);
     _pauseCount++;
   }
-  void resume() { _resume(null); }
+
+  void resume() {
+    _resume(null);
+  }
+
   void _resume(_) {
     if (_pauseCount > 0) _pauseCount--;
   }
-  Future cancel() { return new _Future.immediate(null); }
+
+  Future cancel() {
+    return new _Future.immediate(null);
+  }
+
   bool get isPaused => _pauseCount > 0;
   Future<E> asFuture<E>([E value]) => new _Future<E>();
 }