Add StreamChannel.transform().

lib/stream_channel.dart

 // 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 'dart:convert';
+
+import 'package:async/async.dart';
 
 export 'src/delegating_stream_channel.dart';
 export 'src/multi_channel.dart';
 
 /// An abstract class representing a two-way communication channel.
 ///
 /// Users should consider the [stream] emitting a "done" event to be the
 /// canonical indicator that the channel has closed. If they wish to close the
 /// channel, they should close the [sink]—canceling the stream subscription is
 /// not sufficient. Protocol errors may be emitted through the stream or through
@@ skipping 41 matching lines @@
   /// Creates a new [StreamChannel] that communicates over [stream] and [sink].
   ///
   /// Note that this stream/sink pair must provide the guarantees listed in the
   /// [StreamChannel] documentation.
   factory StreamChannel(Stream<T> stream, StreamSink<T> sink) =>
       new _StreamChannel<T>(stream, sink);
 
   /// 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
 /// as parameters.
 ///
 /// This is distinct from [StreamChannel] so that it can use
 /// [StreamChannelMixin].
 class _StreamChannel<T> extends StreamChannelMixin<T> {
   final Stream<T> stream;
   final StreamSink<T> sink;
 
   _StreamChannel(this.stream, this.sink);
 }
 
 /// A mixin that implements the instance methods of [StreamChannel] in terms of
 /// [stream] and [sink].
 abstract class StreamChannelMixin<T> implements StreamChannel<T> {
   void pipe(StreamChannel<T> other) {
     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));
+  }
 }