Change generic comment syntax to real generic syntax.

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 'package:collection/collection.dart';
 
 import "cancelable_operation.dart";
@@ skipping 113 matching lines @@
   /// one events.
   Future<bool> get hasNext {
     if (!_isClosed) {
       var hasNextRequest = new _HasNextRequest();
       _addRequest(hasNextRequest);
       return hasNextRequest.future;
     }
     throw _failClosed();
   }
 
-
   /// Look at the next [count] data events without consuming them.
   ///
   /// Works like [take] except that the events are left in the queue.
   /// If one of the next [count] events is an error, the returned future
   /// completes with this error, and the error is still left in the queue.
   Future<List<T>> lookAhead(int count) {
     if (count < 0) throw new RangeError.range(count, 0, null, "count");
     if (!_isClosed) {
       var request = new _LookAheadRequest<T>(count);
       _addRequest(request);
@@ skipping 201 matching lines @@
   /// ```dart
   /// final _stdinQueue = new StreamQueue(stdin);
   ///
   /// /// Returns an operation that completes when the user sends a line to
   /// /// standard input.
   /// ///
   /// /// If the operation is canceled, stops waiting for user input.
   /// CancelableOperation<String> nextStdinLine() =>
   ///     _stdinQueue.cancelable((queue) => queue.next);
   /// ```
-  CancelableOperation/*<S>*/ cancelable/*<S>*/(
-      Future/*<S>*/ callback(StreamQueue<T> queue)) {
+  CancelableOperation<S> cancelable<S>(
+      Future<S> callback(StreamQueue<T> queue)) {
     var transaction = startTransaction();
-    var completer = new CancelableCompleter/*<S>*/(onCancel: () {
+    var completer = new CancelableCompleter<S>(onCancel: () {
       transaction.reject();
     });
 
     var queue = transaction.newQueue();
     completer.complete(callback(queue).whenComplete(() {
       if (!completer.isCanceled) transaction.commit(queue);
     }));
 
     return completer.operation;
   }
@@ skipping 117 matching lines @@
   /// immediately, it skips the queue.
   void _addRequest(_EventRequest request) {
     if (_requestQueue.isEmpty) {
       if (request.update(_eventQueue, _isDone)) return;
       _ensureListening();
     }
     _requestQueue.add(request);
   }
 }
 
-
 /// 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<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<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;
   }
 
   void _ensureListening() {
     if (_isDone) return;
     if (_subscription == null) {
-      _subscription =
-          _sourceStream.listen(
-              (data) {
-                _addResult(new Result.value(data));
-              },
-              onError: (error, StackTrace stackTrace) {
-                _addResult(new Result.error(error, stackTrace));
-              },
-              onDone: () {
-                _subscription = null;
-                this._close();
-              });
+      _subscription = _sourceStream.listen((data) {
+        _addResult(new Result.value(data));
+      }, onError: (error, StackTrace stackTrace) {
+        _addResult(new Result.error(error, stackTrace));
+      }, onDone: () {
+        _subscription = null;
+        this._close();
+      });
     } else {
       _subscription.resume();
     }
   }
 
   void _pause() {
     _subscription.pause();
   }
 
   Stream<T> _extractStream() {
@@ skipping 95 matching lines @@
   }
 
   // Cancels all [_queues], removes the [_TransactionRequest] from [_parent]'s
   // request queue, and runs the next request.
   void _done() {
     _splitter.close();
     for (var queue in _queues) {
       queue._cancel();
     }
 
-    assert((_parent._requestQueue.first as _TransactionRequest)
-        .transaction == this);
+    assert((_parent._requestQueue.first as _TransactionRequest).transaction ==
+        this);
     _parent._requestQueue.removeFirst();
     _parent._updateRequests();
   }
 
   /// Throws a [StateError] if [accept] or [reject] has already been called.
   void _assertActive() {
     if (_committed) {
       throw new StateError("This transaction has already been accepted.");
     } else if (_rejected) {
       throw new StateError("This transaction has already been rejected.");
@@ skipping 50 matching lines @@
   _NextRequest();
 
   Future<T> get future => _completer.future;
 
   bool update(QueueList<Result<T>> events, bool isDone) {
     if (events.isNotEmpty) {
       events.removeFirst().complete(_completer);
       return true;
     }
     if (isDone) {
-      _completer.completeError(new StateError("No elements"),
-                               StackTrace.current);
+      _completer.completeError(
+          new StateError("No elements"), StackTrace.current);
       return true;
     }
     return false;
   }
 }
 
-
 /// Request for a [StreamQueue.peek] call.
 ///
 /// Completes the returned future when receiving the first event,
 /// and is then complete, but doesn't consume the event.
 class _PeekRequest<T> implements _EventRequest<T> {
   /// Completer for the future returned by [StreamQueue.next].
   final _completer = new Completer<T>();
 
   _PeekRequest();
 
   Future<T> get future => _completer.future;
 
   bool update(QueueList<Result<T>> events, bool isDone) {
     if (events.isNotEmpty) {
       events.first.complete(_completer);
       return true;
     }
     if (isDone) {
-      _completer.completeError(new StateError("No elements"),
-                               StackTrace.current);
+      _completer.completeError(
+          new StateError("No elements"), StackTrace.current);
       return true;
     }
     return false;
   }
 }
 
-
 /// Request for a [StreamQueue.skip] call.
 class _SkipRequest<T> implements _EventRequest<T> {
   /// Completer for the future returned by the skip call.
   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.
@@ skipping 37 matching lines @@
   /// The request [isComplete] when the length of [_list] reaches
   /// this value.
   final int _eventsToTake;
 
   _ListRequest(this._eventsToTake);
 
   /// The future completed when the correct number of events have been captured.
   Future<List<T>> get future => _completer.future;
 }
 
-
 /// Request for a [StreamQueue.take] call.
 class _TakeRequest<T> extends _ListRequest<T> {
   _TakeRequest(int eventsToTake) : super(eventsToTake);
 
   bool update(QueueList<Result<T>> events, bool isDone) {
     while (_list.length < _eventsToTake) {
       if (events.isEmpty) {
         if (isDone) break;
         return false;
       }
 
       var event = events.removeFirst();
       if (event.isError) {
         event.asError.complete(_completer);
         return true;
       }
       _list.add(event.asValue.value);
     }
     _completer.complete(_list);
     return true;
   }
 }
 
-
 /// Request for a [StreamQueue.lookAhead] call.
 class _LookAheadRequest<T> extends _ListRequest<T> {
   _LookAheadRequest(int eventsToTake) : super(eventsToTake);
 
   bool update(QueueList<Result<T>> events, bool isDone) {
     while (_list.length < _eventsToTake) {
       if (events.length == _list.length) {
         if (isDone) break;
         return false;
       }
       var event = events.elementAt(_list.length);
       if (event.isError) {
         event.asError.complete(_completer);
         return true;
       }
       _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<T> implements _EventRequest<T> {
   /// Completer for the future returned by the `cancel` call.
   final _completer = new Completer();
+
   ///
   /// 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;
 
@@ skipping 35 matching lines @@
       } 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);
+      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.
@@ skipping 39 matching lines @@
   }
 
   bool update(QueueList<Result<T>> events, bool isDone) {
     while (_eventsSent < events.length) {
       events[_eventsSent++].addTo(_controller);
     }
     if (isDone && !_controller.isClosed) _controller.close();
     return false;
   }
 }