Fix most strong mode warnings.

lib/src/stream_queue.dart

 // Copyright (c) 2015, 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.
 
 import 'dart:async';
 import 'dart:collection';
 
 import "result.dart";
 import "subscription_stream.dart";
 import "stream_completer.dart";
@@ skipping 76 matching lines @@
 
   /// Queue of events not used by a request yet.
   final Queue<Result> _eventQueue = new Queue();
 
   /// Queue of pending requests.
   ///
   /// Access through methods below to ensure consistency.
   final Queue<_EventRequest> _requestQueue = new Queue();
 
   /// Create a `StreamQueue` of the events of [source].
-  factory StreamQueue(Stream source) = _StreamQueue<T>;
+  factory StreamQueue(Stream<T> source) = _StreamQueue<T>;
 
   StreamQueue._();
 
   /// Asks if the stream has any more events.
   ///
   /// Returns a future that completes with `true` if the stream has any
   /// more events, whether data or error.
   /// If the stream closes without producing any more events, the returned
   /// future completes with `false`.
   ///
@@ skipping 159 matching lines @@
       _pause();
     }
   }
 
   /// Extracts a stream from the event source and makes this stream queue
   /// unusable.
   ///
   /// Can only be used by the very last request (the stream queue must
   /// be closed by that request).
   /// Only used by [rest].
-  Stream _extractStream();
+  Stream<T> _extractStream();
 
   /// Requests that the event source pauses events.
   ///
   /// This is called automatically when the request queue is empty.
   ///
   /// The event source is restarted by the next call to [_ensureListening].
   void _pause();
 
   /// Ensures that we are listening on events from the event source.
   ///
@@ skipping 47 matching lines @@
   }
 }
 
 
 /// The default implementation of [StreamQueue].
 ///
 /// This queue gets its events from a stream which is listened
 /// to when a request needs events.
 class _StreamQueue<T> extends StreamQueue<T> {
   /// Source of events.
-  final Stream _sourceStream;
+  final Stream<T> _sourceStream;
 
   /// Subscription on [_sourceStream] while listening for events.
   ///
   /// Set to subscription when listening, and set to `null` when the
   /// subscription is done (and [_isDone] is set to true).
-  StreamSubscription _subscription;
+  StreamSubscription<T> _subscription;
 
   _StreamQueue(this._sourceStream) : super._();
 
   Future _cancel() {
     if (_isDone) return null;
     if (_subscription == null) _subscription = _sourceStream.listen(null);
     var future = _subscription.cancel();
     _close();
     return future;
   }
@@ skipping 53 matching lines @@
 ///
 /// Events from the source stream are sent to the first request in the
 /// queue until it reports itself as [isComplete].
 ///
 /// When the first request in the queue `isComplete`, either when becoming
 /// the first request or after receiving an event, its [close] methods is
 /// called.
 ///
 /// The [close] method is also called immediately when the source stream
 /// is done.
-abstract class _EventRequest {
+abstract class _EventRequest<T> {
   /// Handle available events.
   ///
   /// The available events are provided as a queue. The `update` function
   /// should only remove events from the front of the event queue, e.g.,
   /// using [removeFirst].
   ///
   /// Returns `true` if the request is completed, or `false` if it needs
   /// more events.
   /// The call may keep events in the queue until the requeust is complete,
   /// or it may remove them immediately.
   ///
   /// If the method returns true, the request is considered fulfilled, and
   /// will never be called again.
   ///
   /// This method is called when a request reaches the front of the request
   /// queue, and if it returns `false`, it's called again every time a new event
   /// becomes available, or when the stream closes.
   /// If the function returns `false` when the stream has already closed
   /// ([isDone] is true), then the request must call
   /// [StreamQueue._updateRequests] itself when it's ready to continue.
-  bool update(Queue<Result> events, bool isDone);
+  bool update(Queue<Result<T>> events, bool isDone);
 }
 
 /// Request for a [StreamQueue.next] call.
 ///
 /// Completes the returned future when receiving the first event,
 /// and is then complete.
-class _NextRequest<T> implements _EventRequest {
+class _NextRequest<T> implements _EventRequest<T> {
   /// Completer for the future returned by [StreamQueue.next].
-  final Completer _completer;
+  final _completer = new Completer<T>();
 
-  _NextRequest() : _completer = new Completer<T>();
+  _NextRequest();
 
   Future<T> get future => _completer.future;
 
-  bool update(Queue<Result> events, bool isDone) {
+  bool update(Queue<Result<T>> events, bool isDone) {
     if (events.isNotEmpty) {
       events.removeFirst().complete(_completer);
       return true;
     }
     if (isDone) {
       var errorFuture =
           new Future.sync(() => throw new StateError("No elements"));
       _completer.complete(errorFuture);
       return true;
     }
     return false;
   }
 }
 
 /// Request for a [StreamQueue.skip] call.
-class _SkipRequest implements _EventRequest {
+class _SkipRequest<T> implements _EventRequest<T> {
   /// Completer for the future returned by the skip call.
-  final Completer _completer = new Completer<int>();
+  final _completer = new Completer<int>();
 
   /// Number of remaining events to skip.
   ///
   /// The request [isComplete] when the values reaches zero.
   ///
   /// Decremented when an event is seen.
   /// Set to zero when an error is seen since errors abort the skip request.
   int _eventsToSkip;
 
   _SkipRequest(this._eventsToSkip);
 
   /// The future completed when the correct number of events have been skipped.
-  Future get future => _completer.future;
+  Future<int> get future => _completer.future;
 
-  bool update(Queue<Result> events, bool isDone) {
+  bool update(Queue<Result<T>> events, bool isDone) {
     while (_eventsToSkip > 0) {
       if (events.isEmpty) {
         if (isDone) break;
         return false;
       }
       _eventsToSkip--;
 
       var event = events.removeFirst();
       if (event.isError) {
-        event.complete(_completer);
+        _completer.completeError(event.asError.error, event.asError.stackTrace);
         return true;
       }
     }
     _completer.complete(_eventsToSkip);
     return true;
   }
 }
 
 /// Request for a [StreamQueue.take] call.
-class _TakeRequest<T> implements _EventRequest {
+class _TakeRequest<T> implements _EventRequest<T> {
   /// Completer for the future returned by the take call.
-  final Completer _completer;
+  final _completer = new Completer<List<T>>();
 
   /// List collecting events until enough have been seen.
-  final List _list = <T>[];
+  final _list = <T>[];
 
   /// Number of events to capture.
   ///
   /// The request [isComplete] when the length of [_list] reaches
   /// this value.
   final int _eventsToTake;
 
-  _TakeRequest(this._eventsToTake) : _completer = new Completer<List<T>>();
+  _TakeRequest(this._eventsToTake);
 
   /// The future completed when the correct number of events have been captured.
-  Future get future => _completer.future;
+  Future<List<T>> get future => _completer.future;
 
-  bool update(Queue<Result> events, bool isDone) {
+  bool update(Queue<Result<T>> events, bool isDone) {
     while (_list.length < _eventsToTake) {
       if (events.isEmpty) {
         if (isDone) break;
         return false;
       }
 
-      var result = events.removeFirst();
-      if (result.isError) {
-        result.complete(_completer);
+      var event = events.removeFirst();
+      if (event.isError) {
+        _completer.completeError(event.asError.error, event.asError.stackTrace);
         return true;
       }
-      _list.add(result.asValue.value);
+      _list.add(event.asValue.value);
     }
     _completer.complete(_list);
     return true;
   }
 }
 
 /// Request for a [StreamQueue.cancel] call.
 ///
 /// The request needs no events, it just waits in the request queue
 /// until all previous events are fulfilled, then it cancels the stream queue
 /// source subscription.
-class _CancelRequest implements _EventRequest {
+class _CancelRequest<T> implements _EventRequest<T> {
   /// Completer for the future returned by the `cancel` call.
-  final Completer _completer = new Completer();
+  final _completer = new Completer();
 
   /// The [StreamQueue] object that has this request queued.
   ///
   /// When the event is completed, it needs to cancel the active subscription
   /// of the `StreamQueue` object, if any.
   final StreamQueue _streamQueue;
 
   _CancelRequest(this._streamQueue);
 
   /// The future completed when the cancel request is completed.
   Future get future => _completer.future;
 
-  bool update(Queue<Result> events, bool isDone) {
+  bool update(Queue<Result<T>> events, bool isDone) {
     if (_streamQueue._isDone) {
       _completer.complete();
     } else {
       _streamQueue._ensureListening();
       _completer.complete(_streamQueue._extractStream().listen(null).cancel());
     }
     return true;
   }
 }
 
 /// Request for a [StreamQueue.rest] call.
 ///
 /// The request is always complete, it just waits in the request queue
 /// until all previous events are fulfilled, then it takes over the
 /// stream events subscription and creates a stream from it.
-class _RestRequest<T> implements _EventRequest {
+class _RestRequest<T> implements _EventRequest<T> {
   /// Completer for the stream returned by the `rest` call.
-  final StreamCompleter _completer = new StreamCompleter<T>();
+  final _completer = new StreamCompleter<T>();
 
   /// The [StreamQueue] object that has this request queued.
   ///
   /// When the event is completed, it needs to cancel the active subscription
   /// of the `StreamQueue` object, if any.
-  final StreamQueue _streamQueue;
+  final StreamQueue<T> _streamQueue;
 
   _RestRequest(this._streamQueue);
 
   /// The stream which will contain the remaining events of [_streamQueue].
   Stream<T> get stream => _completer.stream;
 
-  bool update(Queue<Result> events, bool isDone) {
+  bool update(Queue<Result<T>> events, bool isDone) {
     if (events.isEmpty) {
       if (_streamQueue._isDone) {
         _completer.setEmpty();
       } else {
         _completer.setSourceStream(_streamQueue._extractStream());
       }
     } else {
       // There are prefetched events which needs to be added before the
       // remaining stream.
       var controller = new StreamController<T>();
       for (var event in events) {
         event.addTo(controller);
       }
       controller.addStream(_streamQueue._extractStream(), cancelOnError: false)
                 .whenComplete(controller.close);
       _completer.setSourceStream(controller.stream);
     }
     return true;
   }
 }
 
 /// Request for a [StreamQueue.hasNext] call.
 ///
 /// Completes the [future] with `true` if it sees any event,
 /// but doesn't consume the event.
 /// If the request is closed without seeing an event, then
 /// the [future] is completed with `false`.
-class _HasNextRequest<T> implements _EventRequest {
-  final Completer _completer = new Completer<bool>();
+class _HasNextRequest<T> implements _EventRequest<T> {
+  final _completer = new Completer<bool>();
 
   Future<bool> get future => _completer.future;
 
-  bool update(Queue<Result> events, bool isDone) {
+  bool update(Queue<Result<T>> events, bool isDone) {
     if (events.isNotEmpty) {
       _completer.complete(true);
       return true;
     }
     if (isDone) {
       _completer.complete(false);
       return true;
     }
     return false;
   }
 }