Add typed wrapper functions to delegate classes.

lib/src/delegate/stream_subscription.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 '../typed/stream_subscription.dart';
+
 /// Simple delegating wrapper around a [StreamSubscription].
 ///
 /// Subclasses can override individual methods.
 class DelegatingStreamSubscription<T> implements StreamSubscription<T> {
   final StreamSubscription _source;
 
   /// Create delegating subscription forwarding calls to [sourceSubscription].
-  DelegatingStreamSubscription(StreamSubscription sourceSubscription)
+  DelegatingStreamSubscription(StreamSubscription<T> sourceSubscription)
       : _source = sourceSubscription;
 
+  /// Creates a wrapper that asserts the types of the values emitted by
+  /// [subscription].
+  ///
+  /// This soundly converts a [StreamSubscription] without a generic type to a
+  /// `StreamSubscription<T>` by asserting that its events are instances of `T`
+  /// whenever they're accessed. If they're not, it throws a [CastError].
+  static StreamSubscription/*<T>*/ typed/*<T>*/(
+          StreamSubscription subscription) =>
+      subscription is StreamSubscription/*<T>*/
+          ? subscription
+          : new TypeSafeStreamSubscription/*<T>*/(subscription);
+
   void onData(void handleData(T data)) {
     _source.onData(handleData);
   }
 
   void onError(Function handleError) {
     _source.onError(handleError);
   }
 
   void onDone(void handleDone()) {
     _source.onDone(handleDone);
   }
 
   void pause([Future resumeFuture]) {
     _source.pause(resumeFuture);
   }
 
   void resume() {
     _source.resume();
   }
 
   Future cancel() => _source.cancel();
 
   Future asFuture([futureValue]) => _source.asFuture(futureValue);
 
   bool get isPaused => _source.isPaused;
 }