Add APIs for asserting the types of transformers.

lib/src/typed_stream_transformer.dart

Index: lib/src/typed_stream_transformer.dart
diff --git a/lib/src/typed_stream_transformer.dart b/lib/src/typed_stream_transformer.dart
new file mode 100644
index 0000000000000000000000000000000000000000..2fb20c468e215816c82e2f5b78247c449e70c9b2
--- /dev/null
+++ b/lib/src/typed_stream_transformer.dart
@@ -0,0 +1,30 @@
+// Copyright (c) 2016, 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 'delegate/stream.dart';
+
+/// Creates a wrapper that coerces the type of [transformer].
+///
+/// This soundly converts a [StreamTransformer] to a `StreamTransformer<S, T>`,
+/// regardless of its original generic type, by asserting that the events
+/// emitted by the transformed stream are instances of `T` whenever they're
+/// provided. If they're not, the stream throws a [CastError].

[Lasse Reichstein Nielsen, 2016/05/09 19:20:12]

Also say that t expects the argument to bind to be a Stream<S> (but it doesn't
actually check that the events are S instances).

[nweiz, 2016/05/09 19:31:51]

This isn't really an aspect of the behavior of the transformer (at least in
strong mode, which is the only mode in which wrapping can actually happen). The
bind() method is annotated as Stream<S>, but that's already statically required
by the fact that this returns a StreamTransformer<S, T>—there's no way for
bind() to actually receive an argument with any other type.

+StreamTransformer/*<S, T>*/ typedStreamTransformer/*<S, T>*/(
+        StreamTransformer transformer) =>
+    transformer is StreamTransformer/*<S, T>*/
+        ? transformer
+        : new _TypeSafeStreamTransformer(transformer);
+
+/// A wrapper that coerces the type of the stream returned by an inner
+/// transformer.
+class _TypeSafeStreamTransformer<S, T> implements StreamTransformer<S, T> {
+  final StreamTransformer _inner;
+
+  _TypeSafeStreamTransformer(this._inner);
+
+  Stream<T> bind(Stream<S> stream) =>
+      DelegatingStream.typed(_inner.bind(stream));
+}