Format all dart: library files

sdk/lib/async/stream_impl.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;
 
 /** Abstract and private interface for a place to put events. */
 abstract class _EventSink<T> {
   void _add(T data);
   void _addError(Object error, StackTrace stackTrace);
@@ skipping 31 matching lines @@
  * * [_close]: Request to close the stream.
  * * [_onCancel]: Called when the subscription will provide no more events,
  *     either due to being actively canceled, or after sending a done event.
  * * [_onPause]: Called when the subscription wants the event source to pause.
  * * [_onResume]: Called when allowing new events after a pause.
  *
  * The user should not add new events when the subscription requests a paused,
  * but if it happens anyway, the subscription will enqueue the events just as
  * when new events arrive while still firing an old event.
  */
-class _BufferingStreamSubscription<T> implements StreamSubscription<T>,
-                                                 _EventSink<T>,
-                                                 _EventDispatch<T> {
+class _BufferingStreamSubscription<T>
+    implements StreamSubscription<T>, _EventSink<T>, _EventDispatch<T> {
   /** The `cancelOnError` flag from the `listen` call. */
   static const int _STATE_CANCEL_ON_ERROR = 1;
   /**
    * Whether the "done" event has been received.
    * No further events are accepted after this.
    */
   static const int _STATE_CLOSED = 2;
   /**
    * Set if the input has been asked not to send events.
    *
@@ skipping 35 matching lines @@
   /** The future [_onCancel] may return. */
   Future _cancelFuture;
 
   /**
    * Queue of pending events.
    *
    * Is created when necessary, or set in constructor for preconfigured events.
    */
   _PendingEvents<T> _pending;
 
-  _BufferingStreamSubscription(void onData(T data),
-                               Function onError,
-                               void onDone(),
-                               bool cancelOnError)
+  _BufferingStreamSubscription(
+      void onData(T data), Function onError, void onDone(), bool cancelOnError)
       : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {
     this.onData(onData);
     this.onError(onError);
     this.onDone(onDone);
   }
 
   /**
    * Sets the subscription's pending events object.
    *
    * This can only be done once. The pending events object is used for the
@@ skipping 66 matching lines @@
     if (!_isCanceled) {
       _cancel();
     }
     return _cancelFuture ?? Future._nullFuture;
   }
 
   Future<E> asFuture<E>([E futureValue]) {
     _Future<E> result = new _Future<E>();
 
     // Overwrite the onDone and onError handlers.
-    _onDone = () { result._complete(futureValue); };
+    _onDone = () {
+      result._complete(futureValue);
+    };
     _onError = (error, stackTrace) {
       Future cancelFuture = cancel();
       if (!identical(cancelFuture, Future._nullFuture)) {
         cancelFuture.whenComplete(() {
           result._completeError(error, stackTrace);
         });
       } else {
         result._completeError(error, stackTrace);
       }
     };
@@ skipping 46 matching lines @@
     if (_canFire) {
       _sendData(data);
     } else {
       _addPending(new _DelayedData<T>(data));
     }
   }
 
   void _addError(Object error, StackTrace stackTrace) {
     if (_isCanceled) return;
     if (_canFire) {
-      _sendError(error, stackTrace);  // Reports cancel after sending.
+      _sendError(error, stackTrace); // Reports cancel after sending.
     } else {
       _addPending(new _DelayedError(error, stackTrace));
     }
   }
 
   void _close() {
     assert(!_isClosed);
     if (_isCanceled) return;
     _state |= _STATE_CLOSED;
     if (_canFire) {
@@ skipping 61 matching lines @@
     assert(!_inCallback);
     bool wasInputPaused = _isInputPaused;
 
     void sendError() {
       // If the subscription has been canceled while waiting for the cancel
       // future to finish we must not report the error.
       if (_isCanceled && !_waitsForCancel) return;
       _state |= _STATE_IN_CALLBACK;
       if (_onError is ZoneBinaryCallback<dynamic, Object, StackTrace>) {
         ZoneBinaryCallback<dynamic, Object, StackTrace> errorCallback = _onError
-            as Object /*=ZoneBinaryCallback<dynamic, Object, StackTrace>*/;
+            as Object/*=ZoneBinaryCallback<dynamic, Object, StackTrace>*/;
         _zone.runBinaryGuarded(errorCallback, error, stackTrace);
       } else {
         _zone.runUnaryGuarded<dynamic, dynamic>(
-            _onError as Object /*=ZoneUnaryCallback<dynamic, dynamic>*/, error);
+            _onError as Object/*=ZoneUnaryCallback<dynamic, dynamic>*/, error);
       }
       _state &= ~_STATE_IN_CALLBACK;
     }
 
     if (_cancelOnError) {
       _state |= _STATE_WAIT_FOR_CANCEL;
       _cancel();
       if (_cancelFuture is Future &&
           !identical(_cancelFuture, Future._nullFuture)) {
         _cancelFuture.whenComplete(sendError);
@@ skipping 90 matching lines @@
 }
 
 // -------------------------------------------------------------------
 // Common base class for single and multi-subscription streams.
 // -------------------------------------------------------------------
 abstract class _StreamImpl<T> extends Stream<T> {
   // ------------------------------------------------------------------
   // Stream interface.
 
   StreamSubscription<T> listen(void onData(T data),
-                               { Function onError,
-                                 void onDone(),
-                                 bool cancelOnError }) {
+      {Function onError, void onDone(), bool cancelOnError}) {
     cancelOnError = identical(true, cancelOnError);
     StreamSubscription<T> subscription =
         _createSubscription(onData, onError, onDone, cancelOnError);
     _onListen(subscription);
     return subscription;
   }
 
   // -------------------------------------------------------------------
   /** Create a subscription object. Called by [subcribe]. */
-  StreamSubscription<T> _createSubscription(
-      void onData(T data),
-      Function onError,
-      void onDone(),
-      bool cancelOnError) {
-    return new _BufferingStreamSubscription<T>(onData, onError, onDone,
-                                               cancelOnError);
+  StreamSubscription<T> _createSubscription(void onData(T data),
+      Function onError, void onDone(), bool cancelOnError) {
+    return new _BufferingStreamSubscription<T>(
+        onData, onError, onDone, cancelOnError);
   }
 
   /** Hook called when the subscription has been created. */
   void _onListen(StreamSubscription subscription) {}
 }
 
 typedef _PendingEvents<T> _EventGenerator<T>();
 
 /** Stream that generates its own events. */
 class _GeneratedStreamImpl<T> extends _StreamImpl<T> {
   final _EventGenerator<T> _pending;
   bool _isUsed = false;
   /**
    * Initializes the stream to have only the events provided by a
    * [_PendingEvents].
    *
    * A new [_PendingEvents] must be generated for each listen.
    */
   _GeneratedStreamImpl(this._pending);
 
-  StreamSubscription<T> _createSubscription(
-      void onData(T data),
-      Function onError,
-      void onDone(),
-      bool cancelOnError) {
+  StreamSubscription<T> _createSubscription(void onData(T data),
+      Function onError, void onDone(), bool cancelOnError) {
     if (_isUsed) throw new StateError("Stream has already been listened to.");
     _isUsed = true;
     return new _BufferingStreamSubscription<T>(
         onData, onError, onDone, cancelOnError).._setPendingEvents(_pending());
   }
 }
 
-
 /** Pending events object that gets its events from an [Iterable]. */
 class _IterablePendingEvents<T> extends _PendingEvents<T> {
   // The iterator providing data for data events.
   // Set to null when iteration has completed.
   Iterator<T> _iterator;
 
   _IterablePendingEvents(Iterable<T> data) : _iterator = data.iterator;
 
   bool get isEmpty => _iterator == null;
 
@@ skipping 21 matching lines @@
       dispatch._sendDone();
     }
   }
 
   void clear() {
     if (isScheduled) cancelSchedule();
     _iterator = null;
   }
 }
 
-
 // Internal helpers.
 
 // Types of the different handlers on a stream. Types used to type fields.
 typedef void _DataHandler<T>(T value);
 typedef void _DoneHandler();
 
-
 /** Default data handler, does nothing. */
 void _nullDataHandler(var value) {}
 
 /** Default error handler, reports the error to the current zone's handler. */
 void _nullErrorHandler(error, [StackTrace stackTrace]) {
   Zone.current.handleUncaughtError(error, stackTrace);
 }
 
 /** Default done handler, does nothing. */
 void _nullDoneHandler() {}
 
-
 /** A delayed event on a buffering stream subscription. */
 abstract class _DelayedEvent<T> {
   /** Added as a linked list on the [StreamController]. */
   _DelayedEvent next;
   /** Execute the delayed event on the [StreamController]. */
   void perform(_EventDispatch<T> dispatch);
 }
 
 /** A delayed data event. */
 class _DelayedData<T> extends _DelayedEvent<T> {
@@ skipping 84 matching lines @@
   void cancelSchedule() {
     if (isScheduled) _state = _STATE_CANCELED;
   }
 
   void handleNext(_EventDispatch<T> dispatch);
 
   /** Throw away any pending events and cancel scheduled events. */
   void clear();
 }
 
-
 /** Class holding pending events for a [_StreamImpl]. */
 class _StreamImplEvents<T> extends _PendingEvents<T> {
   /// Single linked list of [_DelayedEvent] objects.
   _DelayedEvent firstPendingEvent = null;
+
   /// Last element in the list of pending events. New events are added after it.
   _DelayedEvent lastPendingEvent = null;
 
   bool get isEmpty => lastPendingEvent == null;
 
   void add(_DelayedEvent event) {
     if (lastPendingEvent == null) {
       firstPendingEvent = lastPendingEvent = event;
     } else {
       lastPendingEvent = lastPendingEvent.next = event;
@@ skipping 39 matching lines @@
   bool get isPaused => _state >= _PAUSED;
 
   void _schedule() {
     if (_isScheduled) return;
     _zone.scheduleMicrotask(_sendDone);
     _state |= _SCHEDULED;
   }
 
   void onData(void handleData(T data)) {}
   void onError(Function handleError) {}
-  void onDone(void handleDone()) { _onDone = handleDone; }
+  void onDone(void handleDone()) {
+    _onDone = handleDone;
+  }
 
   void pause([Future resumeSignal]) {
     _state += _PAUSED;
     if (resumeSignal != null) resumeSignal.whenComplete(resume);
   }
 
   void resume() {
     if (isPaused) {
       _state -= _PAUSED;
       if (!isPaused && !_isSent) {
         _schedule();
       }
     }
   }
 
   Future cancel() => Future._nullFuture;
 
   Future<E> asFuture<E>([E futureValue]) {
     _Future<E> result = new _Future<E>();
-    _onDone = () { result._completeWithValue(null); };
+    _onDone = () {
+      result._completeWithValue(null);
+    };
     return result;
   }
 
   void _sendDone() {
     _state &= ~_SCHEDULED;
     if (isPaused) return;
     _state |= _DONE_SENT;
     if (_onDone != null) _zone.runGuarded(_onDone);
   }
 }
 
 class _AsBroadcastStream<T> extends Stream<T> {
   final Stream<T> _source;
   final _BroadcastCallback<T> _onListenHandler;
   final _BroadcastCallback<T> _onCancelHandler;
   final Zone _zone;
 
   _AsBroadcastStreamController<T> _controller;
   StreamSubscription<T> _subscription;
 
-  _AsBroadcastStream(this._source,
-                     void onListenHandler(StreamSubscription<T> subscription),
-                     void onCancelHandler(StreamSubscription<T> subscription))
+  _AsBroadcastStream(
+      this._source,
+      void onListenHandler(StreamSubscription<T> subscription),
+      void onCancelHandler(StreamSubscription<T> subscription))
       // TODO(floitsch): the return type should be void and should be
       // inferred.
-      : _onListenHandler = Zone.current.registerUnaryCallback
-            <dynamic, StreamSubscription<T>>(onListenHandler),
-        _onCancelHandler = Zone.current.registerUnaryCallback
-            <dynamic, StreamSubscription<T>>(onCancelHandler),
+      : _onListenHandler = Zone.current
+            .registerUnaryCallback<dynamic, StreamSubscription<T>>(
+                onListenHandler),
+        _onCancelHandler = Zone.current
+            .registerUnaryCallback<dynamic, StreamSubscription<T>>(
+                onCancelHandler),
         _zone = Zone.current {
     _controller = new _AsBroadcastStreamController<T>(_onListen, _onCancel);
   }
 
   bool get isBroadcast => true;
 
   StreamSubscription<T> listen(void onData(T data),
-                               { Function onError,
-                                 void onDone(),
-                                 bool cancelOnError}) {
+      {Function onError, void onDone(), bool cancelOnError}) {
     if (_controller == null || _controller.isClosed) {
       // Return a dummy subscription backed by nothing, since
       // it will only ever send one done event.
       return new _DoneStreamSubscription<T>(onDone);
     }
     if (_subscription == null) {
       _subscription = _source.listen(_controller.add,
-                                     onError: _controller.addError,
-                                     onDone: _controller.close);
+          onError: _controller.addError, onDone: _controller.close);
     }
     cancelOnError = identical(true, cancelOnError);
     return _controller._subscribe(onData, onError, onDone, cancelOnError);
   }
 
   void _onCancel() {
     bool shutdown = (_controller == null) || _controller.isClosed;
     if (_onCancelHandler != null) {
       _zone.runUnary(
           _onCancelHandler, new _BroadcastSubscriptionWrapper<T>(this));
@@ skipping 12 matching lines @@
           _onListenHandler, new _BroadcastSubscriptionWrapper<T>(this));
     }
   }
 
   // Methods called from _BroadcastSubscriptionWrapper.
   void _cancelSubscription() {
     if (_subscription == null) return;
     // Called by [_controller] when it has no subscribers left.
     StreamSubscription subscription = _subscription;
     _subscription = null;
-    _controller = null;  // Marks the stream as no longer listenable.
+    _controller = null; // Marks the stream as no longer listenable.
     subscription.cancel();
   }
 
   void _pauseSubscription(Future resumeSignal) {
     if (_subscription == null) return;
     _subscription.pause(resumeSignal);
   }
 
   void _resumeSubscription() {
     if (_subscription == null) return;
@@ skipping 45 matching lines @@
   bool get isPaused {
     return _stream._isSubscriptionPaused;
   }
 
   Future<E> asFuture<E>([E futureValue]) {
     throw new UnsupportedError(
         "Cannot change handlers of asBroadcastStream source subscription.");
   }
 }
 
-
 /**
  * Simple implementation of [StreamIterator].
  *
  * Pauses the stream between calls to [moveNext].
  */
 class _StreamIterator<T> implements StreamIterator<T> {
   // The stream iterator is always in one of four states.
   // The value of the [_stateData] field depends on the state.
   //
   // When `_subscription == null` and `_stateData != null`:
@@ skipping 39 matching lines @@
   /// Whether the iterator is between calls to `moveNext`.
   /// This will usually cause the [_subscription] to be paused, but as an
   /// optimization, we only pause after the [moveNext] future has been
   /// completed.
   bool _isPaused = false;
 
   _StreamIterator(final Stream<T> stream) : _stateData = stream;
 
   T get current {
     if (_subscription != null && _isPaused) {
-      return _stateData as Object /*=T*/;
+      return _stateData as Object/*=T*/;
     }
     return null;
   }
 
   Future<bool> moveNext() {
     if (_subscription != null) {
       if (_isPaused) {
         var future = new _Future<bool>();
         _stateData = future;
         _isPaused = false;
         _subscription.resume();
         return future;
       }
       throw new StateError("Already waiting for next.");
     }
     return _initializeOrDone();
   }
 
   /// Called if there is no active subscription when [moveNext] is called.
   ///
   /// Either starts listening on the stream if this is the first call to
   /// [moveNext], or returns a `false` future because the stream has already
   /// ended.
   Future<bool> _initializeOrDone() {
     assert(_subscription == null);
     var stateData = _stateData;
     if (stateData != null) {
-      Stream<T> stream = stateData as Object /*=Stream<T>*/;
-      _subscription = stream.listen(
-          _onData, onError: _onError, onDone: _onDone, cancelOnError: true);
+      Stream<T> stream = stateData as Object/*=Stream<T>*/;
+      _subscription = stream.listen(_onData,
+          onError: _onError, onDone: _onDone, cancelOnError: true);
       var future = new _Future<bool>();
       _stateData = future;
       return future;
     }
     return new _Future<bool>.immediate(false);
   }
 
   Future cancel() {
     StreamSubscription<T> subscription = _subscription;
     Object stateData = _stateData;
     _stateData = null;
     if (subscription != null) {
       _subscription = null;
       if (!_isPaused) {
-        _Future<bool> future = stateData as Object /*=_Future<bool>*/;
+        _Future<bool> future = stateData as Object/*=_Future<bool>*/;
         future._asyncComplete(false);
       }
       return subscription.cancel();
     }
     return Future._nullFuture;
   }
 
   void _onData(T data) {
     assert(_subscription != null && !_isPaused);
-    _Future<bool> moveNextFuture = _stateData as Object /*=_Future<bool>*/;
+    _Future<bool> moveNextFuture = _stateData as Object/*=_Future<bool>*/;
     _stateData = data;
     _isPaused = true;
     moveNextFuture._complete(true);
     if (_subscription != null && _isPaused) _subscription.pause();
   }
 
   void _onError(Object error, [StackTrace stackTrace]) {
     assert(_subscription != null && !_isPaused);
-    _Future<bool> moveNextFuture = _stateData as Object /*=_Future<bool>*/;
+    _Future<bool> moveNextFuture = _stateData as Object/*=_Future<bool>*/;
     _subscription = null;
     _stateData = null;
     moveNextFuture._completeError(error, stackTrace);
   }
 
   void _onDone() {
     assert(_subscription != null && !_isPaused);
-    _Future<bool> moveNextFuture = _stateData as Object /*=_Future<bool>*/;
+    _Future<bool> moveNextFuture = _stateData as Object/*=_Future<bool>*/;
     _subscription = null;
     _stateData = null;
     moveNextFuture._complete(false);
   }
 }
 
 /** An empty broadcast stream, sending a done event as soon as possible. */
 class _EmptyStream<T> extends Stream<T> {
   const _EmptyStream() : super._internal();
   bool get isBroadcast => true;
   StreamSubscription<T> listen(void onData(T data),
-                               {Function onError,
-                                void onDone(),
-                                bool cancelOnError}) {
+      {Function onError, void onDone(), bool cancelOnError}) {
     return new _DoneStreamSubscription<T>(onDone);
   }
 }