Add APIs for asserting the types of transformers. I thought about adding these to Delegating*Transformer classes, but those delegates don't really make any sense at all on their own, so I found other places instead. R=lrn@google.com Review URL: https://codereview.chromium.org//1947923003 .

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b18365f..de22933 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md

@@ -1,3 +1,13 @@
+## 1.11.0
+
+* Add a `typedStreamTransformer()` function. This wraps an untyped
+  `StreamTransformer` with the correct type parameters, and asserts the types of
+  events as they're emitted from the transformed stream.
+
+* Add a `StreamSinkTransformer.typed()` static method. This wraps an untyped
+  `StreamSinkTransformer` with the correct type parameters, and asserts the
+  types of arguments passed in to the resulting sink.
+
 ## 1.10.0
 
 * Add `DelegatingFuture.typed()`, `DelegatingStreamSubscription.typed()`,

diff --git a/lib/async.dart b/lib/async.dart
index 7055341..8218986 100644
--- a/lib/async.dart
+++ b/lib/async.dart

@@ -30,3 +30,4 @@
 export "src/stream_splitter.dart";
 export "src/stream_zip.dart";
 export "src/subscription_stream.dart";
+export "src/typed_stream_transformer.dart";

diff --git a/lib/src/stream_sink_transformer.dart b/lib/src/stream_sink_transformer.dart
index 6cde04a..d40bc4b 100644
--- a/lib/src/stream_sink_transformer.dart
+++ b/lib/src/stream_sink_transformer.dart

@@ -6,6 +6,7 @@
 
 import 'stream_sink_transformer/handler_transformer.dart';
 import 'stream_sink_transformer/stream_transformer_wrapper.dart';
+import 'stream_sink_transformer/typed.dart';
 
 /// A [StreamSinkTransformer] transforms the events being passed to a sink.
 ///
@@ -45,4 +46,17 @@
   /// Creates a new sink. When events are passed to the returned sink, it will
   /// transform them and pass the transformed versions to [sink].
   StreamSink<S> bind(StreamSink<T> sink);
+
+  /// Creates a wrapper that coerces the type of [transformer].
+  ///
+  /// This soundly converts a [StreamSinkTransformer] to a
+  /// `StreamSinkTransformer<S, T>`, regardless of its original generic type.
+  /// This means that calls to [StreamSink.add] on the returned sink may throw a
+  /// [CastError] if the argument type doesn't match the reified type of the
+  /// sink.
+  static StreamSinkTransformer/*<S, T>*/ typed/*<S, T>*/(
+          StreamSinkTransformer transformer) =>
+      transformer is StreamSinkTransformer/*<S, T>*/
+          ? transformer
+          : new TypeSafeStreamSinkTransformer(transformer);
 }

diff --git a/lib/src/stream_sink_transformer/typed.dart b/lib/src/stream_sink_transformer/typed.dart
new file mode 100644
index 0000000..303bc08
--- /dev/null
+++ b/lib/src/stream_sink_transformer/typed.dart

@@ -0,0 +1,20 @@
+// 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_sink.dart';
+import '../stream_sink_transformer.dart';
+
+/// A wrapper that coerces the generic type of the sink returned by an inner
+/// transformer to `S`.
+class TypeSafeStreamSinkTransformer<S, T>
+    implements StreamSinkTransformer<S, T> {
+  final StreamSinkTransformer _inner;
+
+  TypeSafeStreamSinkTransformer(this._inner);
+
+  StreamSink<S> bind(StreamSink<T> sink) =>
+      DelegatingStreamSink.typed(_inner.bind(sink));
+}

diff --git a/lib/src/typed_stream_transformer.dart b/lib/src/typed_stream_transformer.dart
new file mode 100644
index 0000000..2fb20c4
--- /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].
+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));
+}

diff --git a/pubspec.yaml b/pubspec.yaml
index 0859e29..c6843a4 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml

@@ -1,5 +1,5 @@
 name: async
-version: 1.10.0
+version: 1.11.0
 author: Dart Team <misc@dartlang.org>
 description: Utility functions and classes related to the 'dart:async' library.
 homepage: https://www.github.com/dart-lang/async