Add StreamChannel.transform().

R=rnystrom@google.com

Review URL: https://codereview.chromium.org//1639643002 .

pkgs/stream_channel/lib/stream_channel.dart

diff --git a/pkgs/stream_channel/lib/stream_channel.dart b/pkgs/stream_channel/lib/stream_channel.dart
index ff36ec7..2885e2d 100644
--- a/pkgs/stream_channel/lib/stream_channel.dart
+++ b/pkgs/stream_channel/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 @@
   /// 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 @@
     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));
+  }
 }

pkgs/stream_channel/test/stream_channel_test.dart

diff --git a/pkgs/stream_channel/test/stream_channel_test.dart b/pkgs/stream_channel/test/stream_channel_test.dart
new file mode 100644
index 0000000..30a7db4
--- /dev/null
+++ b/pkgs/stream_channel/test/stream_channel_test.dart
@@ -0,0 +1,58 @@
+// 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 'dart:isolate';
+
+import 'package:stream_channel/stream_channel.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+void main() {
+  var streamController;
+  var sinkController;
+  var channel;
+  setUp(() {
+    streamController = new StreamController();
+    sinkController = new StreamController();
+    channel = new StreamChannel(
+        streamController.stream, sinkController.sink);
+  });
+
+  test("pipe() pipes data from each channel's stream into the other's sink",
+      () {
+    var otherStreamController = new StreamController();
+    var otherSinkController = new StreamController();
+    var otherChannel = new StreamChannel(
+        otherStreamController.stream, otherSinkController.sink);
+    channel.pipe(otherChannel);
+
+    streamController.add(1);
+    streamController.add(2);
+    streamController.add(3);
+    streamController.close();
+    expect(otherSinkController.stream.toList(), completion(equals([1, 2, 3])));
+
+    otherStreamController.add(4);
+    otherStreamController.add(5);
+    otherStreamController.add(6);
+    otherStreamController.close();
+    expect(sinkController.stream.toList(), completion(equals([4, 5, 6])));
+  });
+
+  test("transform() transforms the channel", () {
+    var transformed = channel.transform(UTF8);
+
+    streamController.add([102, 111, 111, 98, 97, 114]);
+    streamController.close();
+    expect(transformed.stream.toList(), completion(equals(["foobar"])));
+
+    transformed.sink.add("fblthp");
+    transformed.sink.close();
+    expect(sinkController.stream.toList(),
+        completion(equals([[102, 98, 108, 116, 104, 112]])));
+  });
+}