Add cast/retype to Stream, StreamTransformer and Converter.

Switch default for StreamController.addStream cancelOnError parameter to false.
Add orElse named argument to Stream.{first,last}Where.
First step in renaming the argument from defaultValue to orElse.

Change-Id: I22039c1f6933664ebc287c71e802799a64776f08
Reviewed-on: https://dart-review.googlesource.com/34822
Commit-Queue: Lasse R.H. Nielsen <lrn@google.com>
Reviewed-by: Leaf Petersen <leafp@google.com>

pkg/front_end/testcases/shaker/empty_program.dart.outline.expect

@@ -72,6 +72,8 @@ abstract class Stream<T extends core::Object> extends core::Object {
   method any((self::Stream::T) → core::bool test) → self::Future<core::bool>;
   get length() → self::Future<core::int>;
   get isEmpty() → self::Future<core::bool>;
+  method cast<R extends core::Object>() → self::Stream<self::Stream::cast::R>;
+  method retype<R extends core::Object>() → self::Stream<self::Stream::retype::R>;
   method toList() → self::Future<core::List<self::Stream::T>>;
   method toSet() → self::Future<core::Set<self::Stream::T>>;
   method drain<E extends core::Object>([self::Stream::drain::E futureValue]) → self::Future<self::Stream::drain::E>;
@@ -83,9 +85,9 @@ abstract class Stream<T extends core::Object> extends core::Object {
   get first() → self::Future<self::Stream::T>;
   get last() → self::Future<self::Stream::T>;
   get single() → self::Future<self::Stream::T>;
-  method firstWhere((self::Stream::T) → core::bool test, {() → core::Object defaultValue}) → self::Future<dynamic>;
-  method lastWhere((self::Stream::T) → core::bool test, {() → core::Object defaultValue}) → self::Future<dynamic>;
-  method singleWhere((self::Stream::T) → core::bool test) → self::Future<self::Stream::T>;
+  method firstWhere((self::Stream::T) → core::bool test, {() → dynamic defaultValue, () → self::Stream::T orElse}) → self::Future<self::Stream::T>;
+  method lastWhere((self::Stream::T) → core::bool test, {() → dynamic defaultValue, () → self::Stream::T orElse}) → self::Future<self::Stream::T>;
+  method singleWhere((self::Stream::T) → core::bool test, {() → self::Stream::T orElse}) → self::Future<self::Stream::T>;
   method elementAt(core::int index) → self::Future<self::Stream::T>;
   method timeout(core::Duration timeLimit, {(self::EventSink<self::Stream::T>) → void onTimeout}) → self::Stream<self::Stream::T>;
 }
@@ -114,6 +116,8 @@ abstract class StreamSink<S extends core::Object> extends core::Object implement
 }
 abstract class StreamTransformer<S extends core::Object, T extends core::Object> extends core::Object {
   abstract method bind(self::Stream<self::StreamTransformer::S> stream) → self::Stream<self::StreamTransformer::T>;
+  abstract method cast<RS extends core::Object, RT extends core::Object>() → self::StreamTransformer<self::StreamTransformer::cast::RS, self::StreamTransformer::cast::RT>;
+  abstract method retype<RS extends core::Object, RT extends core::Object>() → self::StreamTransformer<self::StreamTransformer::retype::RS, self::StreamTransformer::retype::RT>;
 }
 abstract class StreamIterator<T extends core::Object> extends core::Object {
   abstract method moveNext() → self::Future<core::bool>;

pkg/front_end/testcases/shaker/empty_program.dart.shaker.expect

@@ -62,6 +62,8 @@ library dart:async:
     - any
     - length
     - isEmpty
+    - cast
+    - retype
     - toList
     - toSet
     - drain
@@ -99,6 +101,8 @@ library dart:async:
     - done
   - class StreamTransformer
     - bind
+    - cast
+    - retype
   - class StreamIterator
     - moveNext
     - current

sdk/lib/_http/websocket_impl.dart

@@ -62,11 +62,9 @@ class _CompressionMaxWindowBits {
  * socket will be closed when the processor encounter an error. Not using it
  * will lead to undefined behaviour.
  */
-class _WebSocketProtocolTransformer
-    implements
-        EventSink<List<int>>,
-        StreamTransformer<List<int>,
-            dynamic /*List<int>|_WebSocketPing|_WebSocketPong*/ > {
+class _WebSocketProtocolTransformer extends StreamTransformerBase<List<int>,
+        dynamic /*List<int>|_WebSocketPing|_WebSocketPong*/ >
+    implements EventSink<List<int>> {
   static const int START = 0;
   static const int LEN_FIRST = 1;
   static const int LEN_REST = 2;
@@ -406,7 +404,9 @@ class _WebSocketPong {
 
 typedef /*String|Future<String>*/ _ProtocolSelector(List<String> protocols);
 
-class _WebSocketTransformerImpl implements WebSocketTransformer {
+class _WebSocketTransformerImpl
+    extends StreamTransformerBase<HttpRequest, WebSocket>
+    implements WebSocketTransformer {
   final StreamController<WebSocket> _controller =
       new StreamController<WebSocket>(sync: true);
   final _ProtocolSelector _protocolSelector;
@@ -655,7 +655,7 @@ class _WebSocketPerMessageDeflate {
 
 // TODO(ajohnsen): Make this transformer reusable.
 class _WebSocketOutgoingTransformer
-    implements StreamTransformer<dynamic, List<int>>, EventSink {
+    extends StreamTransformerBase<dynamic, List<int>> implements EventSink {
   final _WebSocketImpl webSocket;
   EventSink<List<int>> _eventSink;
 

sdk/lib/async/async.dart

@@ -92,7 +92,13 @@
 library dart.async;
 
 import "dart:collection" show HashMap, IterableBase;
-import "dart:_internal" show printToZone, printToConsole, IterableElementError;
+import "dart:_internal"
+    show
+        CastStream,
+        CastStreamTransformer,
+        printToZone,
+        printToConsole,
+        IterableElementError;
 
 part 'async_error.dart';
 part 'broadcast_stream_controller.dart';

sdk/lib/async/broadcast_stream_controller.dart

@@ -282,7 +282,7 @@ abstract class _BroadcastStreamController<T>
   Future addStream(Stream<T> stream, {bool cancelOnError}) {
     if (!_mayAddEvent) throw _addEventError();
     _state |= _STATE_ADDSTREAM;
-    _addStreamState = new _AddStreamState(this, stream, cancelOnError ?? true);
+    _addStreamState = new _AddStreamState(this, stream, cancelOnError ?? false);
     return _addStreamState.addStreamFuture;
   }
 

sdk/lib/async/stream.dart

@@ -270,7 +270,7 @@ abstract class Stream<T> {
    *       void close() { _outputSink.close(); }
    *     }
    *
-   *     class DuplicationTransformer implements StreamTransformer<String, String> {
+   *     class DuplicationTransformer extends StreamTransformerBase<String, String> {
    *       // Some generic types omitted for brevity.
    *       Stream bind(Stream stream) => new Stream<String>.eventTransformed(
    *           stream,
@@ -286,6 +286,17 @@ abstract class Stream<T> {
     return new _BoundSinkStream(source, mapSink);
   }
 
+  /**
+   * Adapts [source] to be a `Stream<T>`.
+   *
+   * This allows [source] to be used at the new type, but at run-time it
+   * must satisfy the requirements of both the new type and its original type.
+   *
+   * Data events created by the source stream must also be instances of [T].
+   */
+  static Stream<T> castFrom<S, T>(Stream<S> source) =>
+      new CastStream<S, T>(source);
+
   /**
    * Whether this stream is a broadcast stream.
    */
@@ -921,6 +932,26 @@ abstract class Stream<T> {
     return future;
   }
 
+  /**
+   * Adapt this stream to be a `Stream<R>`.
+   *
+   * If this stream already has the desired type, its returned directly.
+   * Otherwise it is wrapped as a `Stream<R>` which checks at run-time that
+   * each data event emitted by this stream is also an instance of [R].
+   */
+  Stream<R> cast<R>() {
+    Stream<Object> self = this;
+    return self is Stream<R> ? self : retype<R>();
+  }
+
+  /**
+   * Adapt this stream to be a `Stream<R>`.
+   *
+   * This stream is wrapped as a `Stream<R>` which checks at run-time that
+   * each data event emitted by this stream is also an instance of [R].
+   */
+  Stream<R> retype<R>() => Stream.castFrom<T, R>(this);
+
   /**
    * Collects all elements of this stream in a [List].
    *
@@ -1225,8 +1256,8 @@ abstract class Stream<T> {
    * that [test] returns `true` for.
    *
    * If no such element is found before this stream is done, and a
-   * [defaultValue] function is provided, the result of calling [defaultValue]
-   * becomes the value of the future. If [defaultValue] throws, the returned
+   * [orElse] function is provided, the result of calling [orElse]
+   * becomes the value of the future. If [orElse] throws, the returned
    * future is completed with that error.
    *
    * If this stream emits an error before the first matching element,
@@ -1240,11 +1271,12 @@ abstract class Stream<T> {
    * streams are closed and cannot be reused after a call to this method.
    *
    * If an error occurs, or if this stream ends without finding a match and
-   * with no [defaultValue] function provided,
+   * with no [orElse] function provided,
    * the returned future is completed with an error.
    */
-  Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) {
-    _Future<dynamic> future = new _Future();
+  Future<T> firstWhere(bool test(T element),
+      {dynamic defaultValue(), T orElse()}) {
+    _Future<T> future = new _Future();
     StreamSubscription subscription;
     subscription = this.listen(
         (T value) {
@@ -1256,8 +1288,11 @@ abstract class Stream<T> {
         },
         onError: future._completeError,
         onDone: () {
-          if (defaultValue != null) {
-            _runUserCode(defaultValue, future._complete, future._completeError);
+          if (orElse == null && defaultValue != null) {
+            orElse = () => defaultValue() as T;
+          }
+          if (orElse != null) {
+            _runUserCode(orElse, future._complete, future._completeError);
             return;
           }
           try {
@@ -1281,8 +1316,9 @@ abstract class Stream<T> {
    * That means that a non-error result cannot be provided before this stream
    * is done.
    */
-  Future<dynamic> lastWhere(bool test(T element), {Object defaultValue()}) {
-    _Future<dynamic> future = new _Future();
+  Future<T> lastWhere(bool test(T element),
+      {dynamic defaultValue(), T orElse()}) {
+    _Future<T> future = new _Future();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
@@ -1301,8 +1337,11 @@ abstract class Stream<T> {
             future._complete(result);
             return;
           }
-          if (defaultValue != null) {
-            _runUserCode(defaultValue, future._complete, future._completeError);
+          if (orElse == null && defaultValue != null) {
+            orElse = () => defaultValue() as T;
+          }
+          if (orElse != null) {
+            _runUserCode(orElse, future._complete, future._completeError);
             return;
           }
           try {
@@ -1321,7 +1360,7 @@ abstract class Stream<T> {
    * Like [lastWhere], except that it is an error if more than one
    * matching element occurs in the stream.
    */
-  Future<T> singleWhere(bool test(T element)) {
+  Future<T> singleWhere(bool test(T element), {T orElse()}) {
     _Future<T> future = new _Future<T>();
     T result = null;
     bool foundResult = false;
@@ -1350,6 +1389,10 @@ abstract class Stream<T> {
             return;
           }
           try {
+            if (orElse != null) {
+              _runUserCode(orElse, future._complete, future._completeError);
+              return;
+            }
             throw IterableElementError.noElement();
           } catch (e, s) {
             _completeWithErrorCallback(future, e, s);
@@ -1947,6 +1990,21 @@ abstract class StreamTransformer<S, T> {
       void handleError(Object error, StackTrace stackTrace, EventSink<T> sink),
       void handleDone(EventSink<T> sink)}) = _StreamHandlerTransformer<S, T>;
 
+  /**
+   * Adapts [source] to be a `StreamTransfomer<TS, TT>`.
+   *
+   * This allows [source] to be used at the new type, but at run-time it
+   * must satisfy the requirements of both the new type and its original type.
+   *
+   * Data events passed into the returned transformer must also be instances
+   * of [SS], and data events produced by [source] for those events must
+   * also be instances of [TT].
+   */
+  static StreamTransformer<TS, TT> castFrom<SS, ST, TS, TT>(
+      StreamTransformer<SS, ST> source) {
+    return new CastStreamTransformer<SS, ST, TS, TT>(source);
+  }
+
   /**
    * Transforms the provided [stream].
    *
@@ -1969,6 +2027,25 @@ abstract class StreamTransformer<S, T> {
    * duration. Others might not delay them at all, or just by a microtask.
    */
   Stream<T> bind(Stream<S> stream);
+
+  /**
+   * Provides a `StreamTransformer<RS, RT>` view of this stream transformer.
+   *
+   * If this transformer already has the desired type, or a subtype,
+   * it is returned directly,
+   * otherwise returns the result of `retype<RS, RT>()`.
+   */
+  StreamTransformer<RS, RT> cast<RS, RT>();
+
+  /**
+   * Provides a `StreamTrasformer<RS, RT>` view of this stream transformer.
+   *
+   * The resulting transformer will check at run-time that all data events
+   * of the stream it transforms are actually instances of [S],
+   * and it will check that all data events produced by this transformer
+   * are acually instances of [RT].
+   */
+  StreamTransformer<RS, RT> retype<RS, RT>();
 }
 
 /**
@@ -1978,6 +2055,14 @@ abstract class StreamTransformer<S, T> {
  */
 abstract class StreamTransformerBase<S, T> implements StreamTransformer<S, T> {
   const StreamTransformerBase();
+
+  StreamTransformer<RS, RT> cast<RS, RT>() {
+    StreamTransformer<Object, Object> self = this;
+    return self is StreamTransformer<RS, RT> ? self : retype<RS, RT>();
+  }
+
+  StreamTransformer<RS, RT> retype<RS, RT>() =>
+      StreamTransformer.castFrom<S, T, RS, RT>(this);
 }
 
 /**

sdk/lib/async/stream_controller.dart

@@ -271,7 +271,7 @@ abstract class StreamController<T> implements StreamSink<T> {
    * forwarded to the controller's stream, and the `addStream` ends
    * after this. If [cancelOnError] is false, all errors are forwarded
    * and only a done event will end the `addStream`.
-   * If [cancelOnError] is omitted, it defaults to true.
+   * If [cancelOnError] is omitted, it defaults to false.
    */
   Future addStream(Stream<T> source, {bool cancelOnError});
 }
@@ -555,7 +555,7 @@ abstract class _StreamController<T>
     if (_isCanceled) return new _Future.immediate(null);
     _StreamControllerAddStreamState<T> addState =
         new _StreamControllerAddStreamState<T>(
-            this, _varData, source, cancelOnError ?? true);
+            this, _varData, source, cancelOnError ?? false);
     _varData = addState;
     _state |= _STATE_ADDSTREAM;
     return addState.addStreamFuture;
@@ -863,8 +863,9 @@ class _StreamSinkWrapper<T> implements StreamSink<T> {
   }
 
   Future close() => _target.close();
-  Future addStream(Stream<T> source, {bool cancelOnError}) =>
-      _target.addStream(source, cancelOnError: cancelOnError);
+
+  Future addStream(Stream<T> source) => _target.addStream(source);
+
   Future get done => _target.done;
 }
 

sdk/lib/async/stream_transformers.dart

@@ -155,7 +155,7 @@ typedef EventSink<S> _SinkMapper<S, T>(EventSink<T> output);
  *
  * Note that this class can be `const`.
  */
-class _StreamSinkTransformer<S, T> implements StreamTransformer<S, T> {
+class _StreamSinkTransformer<S, T> extends StreamTransformerBase<S, T> {
   final _SinkMapper<S, T> _sinkMapper;
   const _StreamSinkTransformer(this._sinkMapper);
 
@@ -298,7 +298,7 @@ typedef StreamSubscription<T> _SubscriptionTransformer<S, T>(
  * `StreamSubscription`. As such it can also act on `cancel` events, making it
  * fully general.
  */
-class _StreamSubscriptionTransformer<S, T> implements StreamTransformer<S, T> {
+class _StreamSubscriptionTransformer<S, T> extends StreamTransformerBase<S, T> {
   final _SubscriptionTransformer<S, T> _onListen;
 
   const _StreamSubscriptionTransformer(this._onListen);

sdk/lib/convert/convert.dart

@@ -55,7 +55,7 @@ library dart.convert;
 
 import 'dart:async';
 import 'dart:typed_data';
-import 'dart:_internal' show parseHexByte;
+import 'dart:_internal' show CastConverter, parseHexByte;
 
 part 'ascii.dart';
 part 'base64.dart';

sdk/lib/convert/converter.dart

@@ -10,9 +10,21 @@ part of dart.convert;
  * It is recommended that implementations of `Converter` extend this class,
  * to inherit any further methods that may be added to the class.
  */
-abstract class Converter<S, T> implements StreamTransformer<S, T> {
+abstract class Converter<S, T> extends StreamTransformerBase<S, T> {
   const Converter();
 
+  /**
+   * Adapts [source] to be a `Converter<TS, TT>`.
+   *
+   * This allows [source] to be used at the new type, but at run-time it
+   * must satisfy the requirements of both the new type and its original type.
+   *
+   * Conversion input must be both [SS] and [TS] and the output created by
+   * [source] for those input must be both [ST] and [TT].
+   */
+  static Converter<TS, TT> castFrom<SS, ST, TS, TT>(Converter<SS, ST> source) =>
+      new CastConverter<SS, ST, TS, TT>(source);
+
   /**
    * Converts [input] and returns the result of the conversion.
    */
@@ -43,6 +55,28 @@ abstract class Converter<S, T> implements StreamTransformer<S, T> {
     return new Stream<T>.eventTransformed(
         stream, (EventSink sink) => new _ConverterStreamEventSink(this, sink));
   }
+
+  /**
+   * Provides a `Converter<RS, RT>` view of this stream transformer.
+   *
+   * If this transformer already has the desired type, or a subtype,
+   * it is returned directly,
+   * otherwise returns the result of `retype<RS, RT>()`.
+   */
+  Converter<RS, RT> cast<RS, RT>() {
+    Converter<Object, Object> self = this;
+    return self is Converter<RS, RT> ? self : retype<RS, RT>();
+  }
+
+  /**
+   * Provides a `Converter<RS, RT>` view of this stream transformer.
+   *
+   * The resulting transformer will check at run-time that all conversion
+   * inputs are actually instances of [S],
+   * and it will check that all conversion output produced by this converter
+   * are acually instances of [RT].
+   */
+  Converter<RS, RT> retype<RS, RT>() => Converter.castFrom<S, T, RS, RT>(this);
 }
 
 /**

sdk/lib/convert/line_splitter.dart

@@ -18,7 +18,7 @@ const int _CR = 13;
  * The returned lines do not contain the line terminators.
  */
 
-class LineSplitter implements StreamTransformer<String, String> {
+class LineSplitter extends StreamTransformerBase<String, String> {
   const LineSplitter();
 
   /// Split [lines] into individual lines.

sdk/lib/internal/async_cast.dart

@@ -0,0 +1,88 @@
+// 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._internal;
+
+// Casting wrappers for asynchronous classes.
+
+class CastStream<S, T> extends Stream<T> {
+  final Stream<S> _source;
+  CastStream(this._source);
+  bool get isBroadcast => _source.isBroadcast;
+
+  StreamSubscription<T> listen(void onData(T data),
+      {Function onError, void onDone(), bool cancelOnError}) {
+    return new CastStreamSubscription<S, T>(_source.listen(null,
+        onError: onError, onDone: onDone, cancelOnError: cancelOnError))
+      ..onData(onData);
+  }
+
+  Stream<R> cast<R>() {
+    Stream<Object> self = this;
+    return self is Stream<R> ? self : this.retype<R>();
+  }
+
+  Stream<R> retype<R>() => new CastStream<S, R>(_source);
+}
+
+class CastStreamSubscription<S, T> implements StreamSubscription<T> {
+  final StreamSubscription<S> _source;
+
+  CastStreamSubscription(this._source);
+
+  Future cancel() => _source.cancel();
+
+  void onData(void handleData(T data)) {
+    _source.onData((S data) => handleData(data as T));
+  }
+
+  void onError(Function handleError) {
+    _source.onError(handleError);
+  }
+
+  void onDone(void handleDone()) {
+    _source.onDone(handleDone);
+  }
+
+  void pause([Future resumeSignal]) {
+    _source.pause(resumeSignal);
+  }
+
+  void resume() {
+    _source.resume();
+  }
+
+  bool get isPaused => _source.isPaused;
+
+  Future<E> asFuture<E>([E futureValue]) => _source.asFuture<E>(futureValue);
+}
+
+class CastStreamTransformer<SS, ST, TS, TT>
+    extends StreamTransformerBase<TS, TT> {
+  final StreamTransformer<SS, ST> _source;
+  CastStreamTransformer(this._source);
+
+  // cast is inherited from StreamTransformerBase.
+
+  StreamTransformer<RS, RT> retype<RS, RT>() =>
+      new CastStreamTransformer<SS, ST, RS, RT>(_source);
+
+  Stream<TT> bind(Stream<TS> stream) =>
+      _source.bind(stream.cast<SS>()).cast<TT>();
+}
+
+class CastConverter<SS, ST, TS, TT> extends Converter<TS, TT> {
+  final Converter<SS, ST> _source;
+  CastConverter(this._source);
+
+  TT convert(TS input) => _source.convert(input as SS) as TT;
+
+  // cast is inherited from Converter.
+
+  Stream<TT> bind(Stream<TS> stream) =>
+      _source.bind(stream.cast<SS>()).cast<TT>();
+
+  Converter<RS, RT> retype<RS, RT>() =>
+      new CastConverter<SS, ST, RS, RT>(_source);
+}

sdk/lib/internal/cast.dart

@@ -4,7 +4,7 @@
 
 part of dart._internal;
 
-// Casting wrappers for collections and asynchronous classes.
+// Casting wrappers for collection classes.
 
 abstract class _CastIterableBase<S, T> extends Iterable<T> {
   Iterable<S> get _source;

sdk/lib/internal/internal.dart

@@ -6,10 +6,19 @@ library dart._internal;
 
 import 'dart:collection';
 
+import 'dart:async'
+    show
+        Future,
+        Stream,
+        StreamSubscription,
+        StreamTransformer,
+        StreamTransformerBase;
+import 'dart:convert' show Converter;
 import 'dart:core' hide Symbol;
 import 'dart:core' as core;
 import 'dart:math' show Random;
 
+part 'async_cast.dart';
 part 'cast.dart';
 part 'iterable.dart';
 part 'list.dart';

sdk/lib/internal/internal_sources.gni

@@ -7,6 +7,7 @@ internal_sdk_sources = [
   "internal.dart",
 
   # The above file needs to be first as it lists the parts below.
+  "async_cast.dart",
   "cast.dart",
   "iterable.dart",
   "list.dart",

tests/co19/co19-co19.status

@@ -62,6 +62,8 @@ Language/Classes/same_name_type_variable_t07: Pass, MissingCompileTimeError, Fai
 Language/Expressions/Instance_Creation/Const/abstract_class_t01: Pass, Fail # co19 issue 66
 Language/Expressions/Instance_Creation/Const/abstract_class_t03: Pass, Fail # co19 issue 66
 LibTest/async/Stream/asBroadcastStream_A02_t01: Fail # co19 issue 687
+LibTest/async/StreamController/addStream_A03_t01: RuntimeError # Issue <TODO>
+LibTest/async/StreamSink/addStream_A01_t02: RuntimeError # Issue <TODO>
 LibTest/async/Zone/runBinaryGuarded_A01_t01: Fail # co19 issue 126
 LibTest/async/Zone/runGuarded_A01_t01: Fail # co19 issue 126
 LibTest/async/Zone/runUnaryGuarded_A01_t01: Fail # co19 issue 126

tests/lib/lib.status

@@ -166,6 +166,10 @@ async/stream_empty_test: Skip # Flutter Issue 9113
 async/stream_event_transformed_test: Skip # Flutter Issue 9113
 mirrors/*: Skip # Flutter does not support mirrors.
 
+[ $runtime != none ]
+async/stream_controller_async_test: RuntimeError # Library changed.
+async/stream_from_iterable_test: RuntimeError # Library changed.
+
 [ $runtime == safari ]
 convert/json_test: Fail # https://bugs.webkit.org/show_bug.cgi?id=134920
 typed_data/float32x4_test: Fail, Pass # Safari has an optimization bug (nightlies are already fine).
@@ -198,6 +202,14 @@ mirrors/generic_bounded_test/02: Fail # Type equality - Issue 26869
 
 [ $strong ]
 *: SkipByDesign # tests/lib_2 has the strong mode versions of these tests.
+async/stream_controller_async_test: CompileTimeError
+async/stream_first_where_test: CompileTimeError
+async/stream_last_where_test: CompileTimeError
+
+[ !$strong ]
+async/stream_controller_async_test: StaticWarning
+async/stream_first_where_test: RuntimeError
+async/stream_last_where_test: RuntimeError
 
 [ $arch == ia32 && $mode == debug && $system == windows ]
 convert/streamed_conversion_json_utf8_decode_test: Skip # Verification OOM.
@@ -503,6 +515,11 @@ mirrors/deferred_type_test: CompileTimeError, OK # Don't have a multitest marker
 mirrors/native_class_test: Fail, OK # This test is meant to run in a browser.
 typed_data/int32x4_bigint_test: CompileTimeError # Large integer literal
 
+[ $runtime == dart_precompiled || $runtime == flutter || $runtime == vm || $compiler == dart2js && $dart2js_with_kernel ]
+convert/base64_test/01: CompileTimeError # Large integer literal
+convert/utf82_test: CompileTimeError # Large integer literal
+math/double_pow_test: CompileTimeError # Large integer literal
+
 [ $hot_reload || $hot_reload_rollback ]
 async/stream_transformer_test: Pass, Fail # Closure identity
 mirrors/fake_function_with_call_test: SkipByDesign # Method equality

tests/lib_2/async/first_regression_test.dart

@@ -8,7 +8,7 @@ library first_regression_test;
 
 import 'dart:async';
 
-class DoubleTransformer<T> implements StreamTransformer<T, T> {
+class DoubleTransformer<T> extends StreamTransformerBase<T, T> {
   Stream<T> bind(Stream<T> stream) {
     var transformer = new StreamTransformer<T, T>.fromHandlers(
         handleData: (T data, EventSink<T> sink) {

tests/lib_2/async/stream_controller_async_test.dart

@@ -155,8 +155,7 @@ testExtraMethods() {
 
   test("firstWhere 3", () {
     StreamController c = new StreamController();
-    Future f =
-        c.stream.firstWhere((x) => (x % 4) == 0, defaultValue: () => 999);
+    Future f = c.stream.firstWhere((x) => (x % 4) == 0, orElse: () => 999);
     f.then(expectAsync((v) {
       Expect.equals(999, v);
     }));
@@ -181,7 +180,7 @@ testExtraMethods() {
 
   test("lastWhere 3", () {
     StreamController c = new StreamController();
-    Future f = c.stream.lastWhere((x) => (x % 4) == 0, defaultValue: () => 999);
+    Future f = c.stream.lastWhere((x) => (x % 4) == 0, orElse: () => 999);
     f.then(expectAsync((v) {
       Expect.equals(999, v);
     }));
@@ -813,7 +812,7 @@ void testSink({bool sync, bool broadcast, bool asBroadcast}) {
       ..error("BAD")
       ..close();
     StreamController sourceController = new StreamController();
-    c.addStream(sourceController.stream).then((_) {
+    c.addStream(sourceController.stream, cancelOnError: true).then((_) {
       c.close().then((_) {
         Expect.listEquals(expected.events, actual.events);
         done();
@@ -843,7 +842,7 @@ void testSink({bool sync, bool broadcast, bool asBroadcast}) {
       ..close();
 
     StreamController sourceController = new StreamController();
-    c.addStream(sourceController.stream, cancelOnError: false).then((_) {
+    c.addStream(sourceController.stream).then((_) {
       c.close().then((_) {
         Expect.listEquals(source.events, actual.events);
         done();
@@ -881,7 +880,7 @@ void testSink({bool sync, bool broadcast, bool asBroadcast}) {
       ..add(5);
     expected..close();
 
-    c.addStream(s1).then((_) {
+    c.addStream(s1, cancelOnError: true).then((_) {
       c.addStream(s2, cancelOnError: false).then((_) {
         c.close().then((_) {
           Expect.listEquals(expected.events, actual.events);

tests/lib_2/async/stream_event_transformed_test.dart

@@ -71,7 +71,7 @@ class TypeChangingSink implements EventSink<int> {
   }
 }
 
-class SinkTransformer<S, T> implements StreamTransformer<S, T> {
+class SinkTransformer<S, T> extends StreamTransformerBase<S, T> {
   final Function sinkMapper;
   SinkTransformer(this.sinkMapper);
 

tests/lib_2/async/stream_first_where_test.dart

@@ -7,10 +7,7 @@ library stream_controller_async_test;
 import 'dart:async';
 
 import 'package:expect/expect.dart';
-import 'package:unittest/unittest.dart';
-
-import 'event_helper.dart';
-import 'stream_state_helper.dart';
+import 'package:async_helper/async_helper.dart';
 
 class A {
   const A();
@@ -20,17 +17,30 @@ class B extends A {
   const B();
 }
 
-main() {
-  Events sentEvents = new Events()..close();
-
-  // Make sure that firstWhere allows to return instances of types that are
-  // different than the generic type of the stream.
-  test("firstWhere with super class", () {
-    StreamController c = new StreamController<B>();
-    Future f = c.stream.firstWhere((x) => false, defaultValue: () => const A());
-    f.then(expectAsync((v) {
-      Expect.equals(const A(), v);
-    }));
-    sentEvents.replay(c);
-  });
+class C extends B {
+  const C();
+}
+
+main() {
+  asyncStart();
+  {
+    Stream<B> stream = new Stream<B>.fromIterable([new B()]);
+    A aFunc() => const A();
+    // Make sure that firstWhere does not allow you to return instances
+    // of types that are not subtypes of the generic type of the stream.
+    stream.firstWhere((x) => false, //# badType: compile-time error
+        orElse: aFunc); //          //# badType: continued
+  }
+  {
+    asyncStart();
+    C cFunc() => const C();
+    Stream<B> stream = new Stream<B>.fromIterable([new B()]);
+    // Make sure that firstWhere does allow you to return instances
+    // of types that are subtypes of the generic type of the stream.
+    stream.firstWhere((x) => false, orElse: cFunc).then((value) {
+      Expect.identical(const C(), value);
+      asyncEnd();
+    });
+  }
+  asyncEnd();
 }

tests/lib_2/async/stream_from_iterable_test.dart

@@ -105,29 +105,6 @@ main() {
     });
   });
 
-  test("regression-14334-a", () {
-    var from = new Stream.fromIterable([1, 2, 3, 4, 5]);
-
-    // odd numbers as data events, even numbers as error events
-    from = from.map((x) => x.isOdd ? x : throw x);
-
-    var c = new StreamController();
-    var sink = c.sink;
-
-    var done = expectAsync(() {}, count: 2);
-
-    var data = [], errors = [];
-    c.stream.listen(data.add, onError: errors.add, onDone: () {
-      Expect.listEquals([1], data);
-      Expect.listEquals([2], errors);
-      done();
-    });
-    sink.addStream(from).then((_) {
-      c.close();
-      done();
-    });
-  });
-
   test("regression-14334-b", () {
     var from = new Stream.fromIterable([1, 2, 3, 4, 5]);
 
@@ -144,7 +121,7 @@ main() {
       Expect.listEquals([2, 4], errors);
       done();
     });
-    c.addStream(from, cancelOnError: false).then((_) {
+    c.addStream(from).then((_) {
       c.close();
       done();
     });

tests/lib_2/async/stream_last_where_test.dart

@@ -7,10 +7,7 @@ library stream_controller_async_test;
 import 'dart:async';
 
 import 'package:expect/expect.dart';
-import 'package:unittest/unittest.dart';
-
-import 'event_helper.dart';
-import 'stream_state_helper.dart';
+import 'package:async_helper/async_helper.dart';
 
 class A {
   const A();
@@ -20,17 +17,30 @@ class B extends A {
   const B();
 }
 
-main() {
-  Events sentEvents = new Events()..close();
-
-  // Make sure that lastWhere allows to return instances of types that are
-  // different than the generic type of the stream.
-  test("lastWhere with super class", () {
-    StreamController c = new StreamController<B>();
-    Future f = c.stream.lastWhere((x) => false, defaultValue: () => const A());
-    f.then(expectAsync((v) {
-      Expect.equals(const A(), v);
-    }));
-    sentEvents.replay(c);
-  });
+class C extends B {
+  const C();
+}
+
+main() {
+  asyncStart();
+  {
+    Stream<B> stream = new Stream<B>.fromIterable([new B()]);
+    A aFunc() => const A();
+    // Make sure that lastWhere does not allow you to return instances
+    // of types that are not subtypes of the generic type of the stream.
+    stream.lastWhere((x) => false, //# badType: compile-time error
+        orElse: aFunc); //         //# badType: continued
+  }
+  {
+    asyncStart();
+    C cFunc() => const C();
+    Stream<B> stream = new Stream<B>.fromIterable([new B()]);
+    // Make sure that lastWhere does allow you to return instances
+    // of types that are subtypes of the generic type of the stream.
+    stream.lastWhere((x) => false, orElse: cFunc).then((value) {
+      Expect.identical(const C(), value);
+      asyncEnd();
+    });
+  }
+  asyncEnd();
 }

tests/lib_2/lib_2.status

@@ -116,6 +116,10 @@ async/timer_regress22626_test: Pass, RuntimeError # Timing dependent.
 [ $jscl ]
 isolate/spawn_uri_multi_test/none: RuntimeError # Issue 13544
 
+[ !$strong ]
+async/stream_first_where_test/badType: MissingCompileTimeError
+async/stream_last_where_test/badType: MissingCompileTimeError
+
 [ $builder_tag == mac10_7 && $runtime == safari ]
 typed_data/setRange_2_test: Fail # Safari doesn't fully implement spec for TypedArray.set
 typed_data/setRange_3_test: Fail # Safari doesn't fully implement spec for TypedArray.set