Add StreamChannel.transform().

lib/stream_channel.dart

Index: lib/stream_channel.dart
diff --git a/lib/stream_channel.dart b/lib/stream_channel.dart
index ff36ec78bd404cb1799875c3d031a59694cbaef4..2885e2d23780eed7587478cb1b8925dc387a7cb7 100644
--- a/lib/stream_channel.dart
+++ b/lib/stream_channel.dart
@@ -3,6 +3,9 @@
 // BSD-style license that can be found in the LICENSE file.
 
 import 'dart:async';
+import 'dart:convert';
+
+import 'package:async/async.dart';
 
 export 'src/delegating_stream_channel.dart';
 export 'src/isolate_channel.dart';
@@ -66,6 +69,13 @@ abstract class StreamChannel<T> {
   /// Connects [this] to [other], so that any values emitted by either are sent
   /// directly to the other.
   void pipe(StreamChannel<T> other);
+
+  /// Transforms [this] using [codec].
+  ///
+  /// 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);
 }
 
 /// An implementation of [StreamChannel] that simply takes a stream and a sink
@@ -87,4 +97,12 @@ abstract class StreamChannelMixin<T> implements StreamChannel<T> {
     stream.pipe(other.sink);
     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));
+  }
 }