Fix all strong-mode errors and warnings.

lib/src/util/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.
 
 // TODO(nweiz): Get rid of this when https://codereview.chromium.org/1241723003/
 // lands.
 import 'dart:async';
 import 'dart:collection';
 
 import "package:async/async.dart" hide ForkableStream, StreamQueue;
@@ skipping 60 matching lines @@
   // again when new events are available.
   // The request can remove events that it uses, or keep them in the event
   // queue until it has all that it needs.
   //
   // This model is very flexible and easily extensible.
   // It allows requests that don't consume events (like [hasNext]) or
   // potentially a request that takes either five or zero events, determined
   // by the content of the fifth event.
 
   /// Source of events.
-  final ForkableStream _sourceStream;
+  final ForkableStream<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;
 
   /// Whether we have listened on [_sourceStream] and the subscription is done.
   bool _isDone = false;
 
   /// Whether a closing operation has been performed on the stream queue.
   ///
   /// Closing operations are [cancel] and [rest].
   bool _isClosed = false;
 
   /// 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].
-  StreamQueue(Stream source)
+  StreamQueue(Stream<T> source)
       : _sourceStream = source is ForkableStream
           ? source
           : new ForkableStream(source);
 
   /// 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 237 matching lines @@
     }
 
     if (!_isDone) {
       _subscription.pause();
     }
   }
 
   /// Extracts the subscription and makes this stream queue unusable.
   ///
   /// Can only be used by the very last request.
-  StreamSubscription _dispose() {
+  StreamSubscription<T> _dispose() {
     assert(_isClosed);
     var subscription = _subscription;
     _subscription = null;
     _isDone = true;
     return subscription;
   }
 }
 
 /// Request object that receives events when they arrive, until fulfilled.
 ///
@@ skipping 44 matching lines @@
   /// this is the last chance to use them.
   void close(Queue<Result> events);
 }
 
 /// Request for a [StreamQueue.next] call.
 ///
 /// Completes the returned future when receiving the first event,
 /// and is then complete.
 class _NextRequest<T> implements _EventRequest {
   /// 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 addEvents(Queue<Result> events) {
     if (events.isEmpty) return false;
     events.removeFirst().complete(_completer);
     return true;
   }
 
   void close(Queue<Result> events) {
     var errorFuture =
         new Future.sync(() => throw new StateError("No elements"));
     _completer.complete(errorFuture);
   }
 }
 
 /// Request for a [StreamQueue.skip] call.
 class _SkipRequest implements _EventRequest {
   /// 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 addEvents(Queue<Result> events) {
     while (_eventsToSkip > 0) {
       if (events.isEmpty) return false;
       _eventsToSkip--;
       var event = events.removeFirst();
       if (event.isError) {
         event.complete(_completer);
         return true;
       }
     }
     _completer.complete(0);
     return true;
   }
 
   void close(Queue<Result> events) {
     _completer.complete(_eventsToSkip);
   }
 }
 
 /// Request for a [StreamQueue.take] call.
 class _TakeRequest<T> implements _EventRequest {
   /// 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>[];
 
   /// 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 addEvents(Queue<Result> events) {
     while (_list.length < _eventsToTake) {
       if (events.isEmpty) return false;
       var result = events.removeFirst();
       if (result.isError) {
         result.complete(_completer);
         return true;
       }
       _list.add(result.asValue.value);
@@ skipping 46 matching lines @@
   }
 }
 
 /// 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 {
   /// 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 addEvents(Queue<Result> events) {
     _completeStream(events);
     return true;
   }
@@ skipping 16 matching lines @@
       for (var event in events) {
         event.addTo(controller);
       }
       controller.addStream(_getRestStream(), cancelOnError: false)
                 .whenComplete(controller.close);
       _completer.setSourceStream(controller.stream);
     }
   }
 
   /// Create a stream from the rest of [_streamQueue]'s subscription.
-  Stream _getRestStream() {
+  Stream<T> _getRestStream() {
     if (_streamQueue._isDone) {
       var controller = new StreamController<T>()..close();
       return controller.stream;
       // TODO(lrn). Use the following when 1.11 is released.
       // return new Stream<T>.empty();
     }
     if (_streamQueue._subscription == null) {
       return _streamQueue._sourceStream;
     }
     var subscription = _streamQueue._dispose();
     subscription.resume();
     return new SubscriptionStream<T>(subscription);
   }
 }
 
 /// 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>();
+  final _completer = new Completer<bool>();
 
   Future<bool> get future => _completer.future;
 
   bool addEvents(Queue<Result> events) {
     if (events.isNotEmpty) {
       _completer.complete(true);
       return true;
     }
     return false;
   }
 
   void close(_) {
     _completer.complete(false);
   }
 }
 
 /// Request for a [StreamQueue.fork] call.
 class _ForkRequest<T> implements _EventRequest {
   /// Completer for the stream used by the queue by the `fork` call.
-  StreamCompleter _completer;
+  StreamCompleter<T> _completer;
 
   StreamQueue<T> queue;
 
   /// The [StreamQueue] object that has this request queued.
-  final StreamQueue _streamQueue;
+  final StreamQueue<T> _streamQueue;
 
   _ForkRequest(this._streamQueue) {
-    _completer = new StreamCompleter<T>();
-    queue = new StreamQueue<T>(_completer.stream);
+    _completer = new StreamCompleter();
+    queue = new StreamQueue(_completer.stream);
   }
 
   bool addEvents(Queue<Result> events) {
     _completeStream(events);
     return true;
   }
 
   void close(Queue<Result> events) {
     _completeStream(events);
   }
@@ skipping 13 matching lines @@
         event.addTo(controller);
       }
 
       var fork = _streamQueue._sourceStream.fork();
       controller.addStream(fork, cancelOnError: false)
           .whenComplete(controller.close);
       _completer.setSourceStream(controller.stream);
     }
   }
 }