Add StreamChannelTransformer.

lib/stream_channel.dart

Index: lib/stream_channel.dart
diff --git a/lib/stream_channel.dart b/lib/stream_channel.dart
index 2885e2d23780eed7587478cb1b8925dc387a7cb7..e7fb055aa0547b96624b0ffae95b4dc57a1392d8 100644
--- a/lib/stream_channel.dart
+++ b/lib/stream_channel.dart
@@ -3,14 +3,14 @@
 // BSD-style license that can be found in the LICENSE file.
 
 import 'dart:async';
-import 'dart:convert';
 
-import 'package:async/async.dart';
+import 'src/stream_channel_transformer.dart';
 
 export 'src/delegating_stream_channel.dart';
 export 'src/isolate_channel.dart';
 export 'src/multi_channel.dart';
 export 'src/stream_channel_completer.dart';
+export 'src/stream_channel_transformer.dart';
 
 /// An abstract class representing a two-way communication channel.
 ///
@@ -70,12 +70,10 @@ abstract class StreamChannel<T> {
   /// directly to the other.
   void pipe(StreamChannel<T> other);
 
-  /// Transforms [this] using [codec].
+  /// Transforms [this] using [transformer].
   ///
-  /// This returns a stream channel that encodes all input using [Codec.encoder]
-  /// before passing it to this channel's [sink], and decodes all output from
-  /// this channel's [stream] using [Codec.decoder].
-  StreamChannel transform(Codec<dynamic, T> codec);
+  /// This is identical to calling `transformer.bind(channel)`.
+  StreamChannel transform(StreamChannelTransformer<dynamic, T> transformer);
 }
 
 /// An implementation of [StreamChannel] that simply takes a stream and a sink
@@ -98,11 +96,6 @@ abstract class StreamChannelMixin<T> implements StreamChannel<T> {
     other.stream.pipe(sink);
   }
 
-  StreamChannel transform(Codec<dynamic, T> codec) {
-    var sinkTransformer =
-        new StreamSinkTransformer.fromStreamTransformer(codec.encoder);
-    return new _StreamChannel(
-        stream.transform(codec.decoder),
-        sinkTransformer.bind(sink));
-  }
+  StreamChannel transform(StreamChannelTransformer<dynamic, T> transformer) =>
+      transformer.bind(this);
 }