Add StreamChannel.withCloseGuarantee.

test/with_close_guarantee_test.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 'package:async/async.dart';
+import 'package:stream_channel/stream_channel.dart';
+import 'package:test/test.dart';
+
+import 'utils.dart';
+
+final _delayTransformer = new StreamTransformer.fromHandlers(
+    handleData: (data, sink) => new Future.microtask(() => sink.add(data)),
+    handleDone: (sink) => new Future.microtask(() => sink.close()));
+
+final _delaySinkTransformer =
+    new StreamSinkTransformer.fromStreamTransformer(_delayTransformer);
+
+void main() {
+  var controller;
+  var channel;
+  setUp(() {
+    controller = new StreamChannelController();
+
+    // Add a bunch of layers of asynchronous dispatch between the channel and
+    // the underlying controllers.
+    var stream = controller.foreign.stream;
+    var sink = controller.foreign.sink;
+    for (var i = 0; i < 10; i++) {
+      stream = stream.transform(_delayTransformer);
+      sink = _delaySinkTransformer.bind(sink);
+    }
+
+    channel = new StreamChannel.withCloseGuarantee(stream, sink);
+  });
+
+  test("closing the event sink causes the stream to close before it emits any "
+      "more events", () async {
+    controller.local.sink.add(1);
+    controller.local.sink.add(2);
+    controller.local.sink.add(3);
+
+    expect(channel.stream.listen(expectAsync((event) {
+      if (event == 2) channel.sink.close();
+    }, count: 2)).asFuture(), completes);
+
+    await pumpEventQueue();
+  });
+
+  test("closing the event sink before events are emitted causes the stream to "
+      "close immediately", () async {
+    channel.sink.close();
+    channel.stream.listen(
+        expectAsync((_) {}, count: 0),
+        onError: expectAsync((_, __) {}, count: 0),
+        onDone: expectAsync(() {}));
+
+    controller.local.sink.add(1);
+    controller.local.sink.add(2);
+    controller.local.sink.add(3);
+    controller.local.sink.close();
+
+    await pumpEventQueue();
+  });
+}