diff --git a/CHANGELOG.md b/CHANGELOG.md
index af87e24..15570d7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,12 @@
 - Added `StreamSinkCompleter`, for creating a `StreamSink` now and providing its
   destination later as another sink.
 
+- Added `StreamCompleter.setError`, a shortcut for emitting a single error event
+  on the resulting stream.
+
+- Added `NullStreamSink`, an implementation of `StreamSink` that discards all
+  events.
+
 ## 1.7.0
 
 - Added `SingleSubscriptionTransformer`, a `StreamTransformer` that converts a
diff --git a/lib/async.dart b/lib/async.dart
index ef59b73..ef1ac04 100644
--- a/lib/async.dart
+++ b/lib/async.dart
@@ -15,6 +15,7 @@
 export "src/delegate/stream_subscription.dart";
 export "src/future_group.dart";
 export "src/lazy_stream.dart";
+export "src/null_stream_sink.dart";
 export "src/restartable_timer.dart";
 export "src/result_future.dart";
 export "src/single_subscription_transformer.dart";
diff --git a/lib/src/null_stream_sink.dart b/lib/src/null_stream_sink.dart
new file mode 100644
index 0000000..aa85924
--- /dev/null
+++ b/lib/src/null_stream_sink.dart
@@ -0,0 +1,90 @@
+// 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.
+
+library async.null_stream_sink;
+
+import 'dart:async';
+
+/// A [StreamSink] that discards all events.
+///
+/// The sink silently drops events until [close] is called, at which point it
+/// throws [StateError]s when events are added. This is the same behavior as a
+/// sink whose remote end has closed, such as when a [WebSocket] connection has
+/// been closed.
+///
+/// This can be used when a sink is needed but no events are actually intended
+/// to be added. The [new NullStreamSink.error] constructor can be used to
+/// represent errors when creating a sink, since [StreamSink.done] exposes sink
+/// errors. For example:
+///
+/// ```dart
+/// StreamSink<List<int>> openForWrite(String filename) {
+///   try {
+///     return new RandomAccessSink(new File(filename).openSync());
+///   } on IOException catch (error, stackTrace) {
+///     return new NullStreamSink.error(error, stackTrace);
+///   }
+/// }
+/// ```
+class NullStreamSink<T> implements StreamSink<T> {
+  final Future done;
+
+  /// Whether the sink has been closed.
+  var _closed = false;
+
+  /// Whether an [addStream] call is pending.
+  ///
+  /// We don't actually add any events from streams, but it does return the
+  /// [StreamSubscription.cancel] future so to be [StreamSink]-complaint we
+  /// reject events until that completes.
+  var _addingStream = false;
+
+  /// Creates a null sink.
+  ///
+  /// If [done] is passed, it's used as the [Sink.done] future. Otherwise, a
+  /// completed future is used.
+  NullStreamSink({Future done}) : done = done ?? new Future.value();
+
+  /// Creates a null sink whose [done] future emits [error].
+  ///
+  /// Note that this error will not be considered uncaught.
+  NullStreamSink.error(error, [StackTrace stackTrace])
+      : done = new Future.error(error, stackTrace)
+          // Don't top-level the error. This gives the user a change to call
+          // [close] or [done], and matches the behavior of a remote endpoint
+          // experiencing an error.
+          ..catchError((_) {});
+
+  void add(T data) {
+    _checkEventAllowed();
+  }
+
+  void addError(error, [StackTrace stackTrace]) {
+    _checkEventAllowed();
+  }
+
+  Future addStream(Stream<T> stream) {
+    _checkEventAllowed();
+
+    _addingStream = true;
+    var future = stream.listen(null).cancel() ?? new Future.value();
+    return future.whenComplete(() {
+      _addingStream = false;
+    });
+  }
+
+  /// Throws a [StateError] if [close] has been called or an [addStream] call is
+  /// pending.
+  void _checkEventAllowed() {
+    if (_closed) throw new StateError("Cannot add to a closed sink.");
+    if (_addingStream) {
+      throw new StateError("Cannot add to a sink while adding a stream.");
+    }
+  }
+
+  Future close() {
+    _closed = true;
+    return done;
+  }
+}
diff --git a/lib/src/stream_completer.dart b/lib/src/stream_completer.dart
index c343e6e..a6260dc 100644
--- a/lib/src/stream_completer.dart
+++ b/lib/src/stream_completer.dart
@@ -39,9 +39,7 @@
   static Stream fromFuture(Future<Stream> streamFuture) {
     var completer = new StreamCompleter();
     streamFuture.then(completer.setSourceStream,
-                      onError: (e, s) {
-                        completer.setSourceStream(streamFuture.asStream());
-                      });
+        onError: completer.setError);
     return completer.stream;
   }
 
@@ -76,8 +74,8 @@
   /// it is immediately listened to, and its events are forwarded to the
   /// existing subscription.
   ///
-  /// Either [setSourceStream] or [setEmpty] may be called at most once.
-  /// Trying to call either of them again will fail.
+  /// Any one of [setSourceStream], [setEmpty], and [setError] may be called at
+  /// most once. Trying to call any of them again will fail.
   void setSourceStream(Stream<T> sourceStream) {
     if (_stream._isSourceStreamSet) {
       throw new StateError("Source stream already set");
@@ -87,14 +85,24 @@
 
   /// Equivalent to setting an empty stream using [setSourceStream].
   ///
-  /// Either [setSourceStream] or [setEmpty] may be called at most once.
-  /// Trying to call either of them again will fail.
+  /// Any one of [setSourceStream], [setEmpty], and [setError] may be called at
+  /// most once. Trying to call any of them again will fail.
   void setEmpty() {
     if (_stream._isSourceStreamSet) {
       throw new StateError("Source stream already set");
     }
     _stream._setEmpty();
   }
+
+  /// Completes this to a stream that emits [error] and then closes.
+  ///
+  /// This is useful when the process of creating the data for the stream fails.
+  ///
+  /// Any one of [setSourceStream], [setEmpty], and [setError] may be called at
+  /// most once. Trying to call any of them again will fail.
+  void setError(error, [StackTrace stackTrace]) {
+    setSourceStream(new Stream.fromFuture(new Future.error(error, stackTrace)));
+  }
 }
 
 /// Stream completed by [StreamCompleter].
diff --git a/lib/src/stream_sink_completer.dart b/lib/src/stream_sink_completer.dart
index 10caa06..0b6dc46 100644
--- a/lib/src/stream_sink_completer.dart
+++ b/lib/src/stream_sink_completer.dart
@@ -6,6 +6,8 @@
 
 import 'dart:async';
 
+import 'null_stream_sink.dart';
+
 /// A [sink] where the destination is provided later.
 ///
 /// The [sink] is a normal sink that you can add events to to immediately, but
@@ -29,6 +31,20 @@
   /// Returns [sink] typed as a [_CompleterSink].
   _CompleterSink<T> get _sink => sink;
 
+  /// Convert a `Future<StreamSink>` to a `StreamSink`.
+  ///
+  /// This creates a sink using a sink completer, and sets the destination sink
+  /// to the result of the future when the future completes.
+  ///
+  /// If the future completes with an error, the returned sink will instead
+  /// be closed. Its [Sink.done] future will contain the error.
+  static StreamSink fromFuture(Future<StreamSink> sinkFuture) {
+    var completer = new StreamSinkCompleter();
+    sinkFuture.then(completer.setDestinationSink,
+        onError: completer.setError);
+    return completer.sink;
+  }
+
   /// Sets a sink as the destination for events from the [StreamSinkCompleter]'s
   /// [sink].
   ///
@@ -41,12 +57,25 @@
   /// buffered until the destination is available.
   ///
   /// A destination sink may be set at most once.
+  ///
+  /// Either of [setDestinationSink] or [setError] may be called at most once.
+  /// Trying to call either of them again will fail.
   void setDestinationSink(StreamSink<T> destinationSink) {
     if (_sink._destinationSink != null) {
       throw new StateError("Destination sink already set");
     }
     _sink._setDestinationSink(destinationSink);
   }
+
+  /// Completes this to a closed sink whose [Sink.done] future emits [error].
+  ///
+  /// This is useful when the process of loading the sink fails.
+  ///
+  /// Either of [setDestinationSink] or [setError] may be called at most once.
+  /// Trying to call either of them again will fail.
+  void setError(error, [StackTrace stackTrace]) {
+    setDestinationSink(new NullStreamSink.error(error, stackTrace));
+  }
 }
 
 /// [StreamSink] completed by [StreamSinkCompleter].
diff --git a/pubspec.yaml b/pubspec.yaml
index ccfa970..7560836 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -1,5 +1,5 @@
 name: async
-version: 1.8.0-dev
+version: 1.8.0
 author: Dart Team <misc@dartlang.org>
 description: Utility functions and classes related to the 'dart:async' library.
 homepage: https://www.github.com/dart-lang/async
diff --git a/test/null_stream_sink_test.dart b/test/null_stream_sink_test.dart
new file mode 100644
index 0000000..244f99c
--- /dev/null
+++ b/test/null_stream_sink_test.dart
@@ -0,0 +1,113 @@
+// 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:test/test.dart";
+
+import "utils.dart";
+
+void main() {
+  group("constructors", () {
+    test("done defaults to a completed future", () {
+      var sink = new NullStreamSink();
+      expect(sink.done, completes);
+    });
+
+    test("a custom future may be passed to done", () async {
+      var completer = new Completer();
+      var sink = new NullStreamSink(done: completer.future);
+
+      var doneFired = false;
+      sink.done.then((_) {
+        doneFired = true;
+      });
+      await flushMicrotasks();
+      expect(doneFired, isFalse);
+
+      completer.complete();
+      await flushMicrotasks();
+      expect(doneFired, isTrue);
+    });
+
+    test("NullStreamSink.error passes an error to done", () {
+      var sink = new NullStreamSink.error("oh no");
+      expect(sink.done, throwsA("oh no"));
+    });
+  });
+
+  group("events", () {
+    test("are silently dropped before close", () {
+      var sink = new NullStreamSink();
+      sink.add(1);
+      sink.addError("oh no");
+    });
+
+    test("throw StateErrors after close", () {
+      var sink = new NullStreamSink();
+      expect(sink.close(), completes);
+
+      expect(() => sink.add(1), throwsStateError);
+      expect(() => sink.addError("oh no"), throwsStateError);
+      expect(() => sink.addStream(new Stream.empty()), throwsStateError);
+    });
+
+    group("addStream", () {
+      test("listens to the stream then cancels immediately", () async {
+        var sink = new NullStreamSink();
+        var canceled = false;
+        var controller = new StreamController(onCancel: () {
+          canceled = true;
+        });
+
+        expect(sink.addStream(controller.stream), completes);
+        await flushMicrotasks();
+        expect(canceled, isTrue);
+      });
+
+      test("returns the cancel future", () async {
+        var completer = new Completer();
+        var sink = new NullStreamSink();
+        var controller = new StreamController(onCancel: () => completer.future);
+
+        var addStreamFired = false;
+        sink.addStream(controller.stream).then((_) {
+          addStreamFired = true;
+        });
+        await flushMicrotasks();
+        expect(addStreamFired, isFalse);
+
+        completer.complete();
+        await flushMicrotasks();
+        expect(addStreamFired, isTrue);
+      });
+
+      test("pipes errors from the cancel future through addStream", () async {
+        var sink = new NullStreamSink();
+        var controller = new StreamController(onCancel: () => throw "oh no");
+        expect(sink.addStream(controller.stream), throwsA("oh no"));
+      });
+
+      test("causes events to throw StateErrors until the future completes",
+          () async {
+        var sink = new NullStreamSink();
+        var future = sink.addStream(new Stream.empty());
+        expect(() => sink.add(1), throwsStateError);
+        expect(() => sink.addError("oh no"), throwsStateError);
+        expect(() => sink.addStream(new Stream.empty()), throwsStateError);
+
+        await future;
+        sink.add(1);
+        sink.addError("oh no");
+        expect(sink.addStream(new Stream.empty()), completes);
+      });
+    });
+  });
+
+  test("close returns the done future", () {
+    var sink = new NullStreamSink.error("oh no");
+    expect(sink.close(), throwsA("oh no"));
+  });
+}
diff --git a/test/stream_completer_test.dart b/test/stream_completer_test.dart
index cd3ceb9..76d81d1 100644
--- a/test/stream_completer_test.dart
+++ b/test/stream_completer_test.dart
@@ -337,6 +337,34 @@
     });
     expect(controller.hasListener, isFalse);
   });
+
+  group("setError()", () {
+    test("produces a stream that emits a single error", () {
+      var completer = new StreamCompleter();
+      completer.stream.listen(
+          unreachable("data"),
+          onError: expectAsync((error, stackTrace) {
+            expect(error, equals("oh no"));
+          }),
+          onDone: expectAsync(() {}));
+
+      completer.setError("oh no");
+    });
+
+    test("produces a stream that emits a single error on a later listen",
+        () async {
+      var completer = new StreamCompleter();
+      completer.setError("oh no");
+      await flushMicrotasks();
+
+      completer.stream.listen(
+          unreachable("data"),
+          onError: expectAsync((error, stackTrace) {
+            expect(error, equals("oh no"));
+          }),
+          onDone: expectAsync(() {}));
+    });
+  });
 }
 
 Stream<int> createStream() async* {
diff --git a/test/stream_sink_completer_test.dart b/test/stream_sink_completer_test.dart
index 3892d26..82393fc 100644
--- a/test/stream_sink_completer_test.dart
+++ b/test/stream_sink_completer_test.dart
@@ -245,6 +245,48 @@
     expect(completer.sink.close(), completes);
   });
 
+  group("fromFuture()", () {
+    test("with a successful completion", () async {
+      var futureCompleter = new Completer();
+      var sink = StreamSinkCompleter.fromFuture(futureCompleter.future);
+      sink.add(1);
+      sink.add(2);
+      sink.add(3);
+      sink.close();
+
+      var testSink = new TestSink();
+      futureCompleter.complete(testSink);
+      await testSink.done;
+
+      expect(testSink.results[0].asValue.value, equals(1));
+      expect(testSink.results[1].asValue.value, equals(2));
+      expect(testSink.results[2].asValue.value, equals(3));
+    });
+
+    test("with an error", () async {
+      var futureCompleter = new Completer();
+      var sink = StreamSinkCompleter.fromFuture(futureCompleter.future);
+      expect(sink.done, throwsA("oh no"));
+      futureCompleter.completeError("oh no");
+    });
+  });
+
+  group("setError()", () {
+    test("produces a closed sink with the error", () {
+      completer.setError("oh no");
+      expect(completer.sink.done, throwsA("oh no"));
+      expect(completer.sink.close(), throwsA("oh no"));
+    });
+
+    test("produces an error even if done was accessed earlier", () async {
+      expect(completer.sink.done, throwsA("oh no"));
+      expect(completer.sink.close(), throwsA("oh no"));
+      await flushMicrotasks();
+
+      completer.setError("oh no");
+    });
+  });
+
   test("doesn't allow the destination sink to be set multiple times", () {
     completer.setDestinationSink(new TestSink());
     expect(() => completer.setDestinationSink(new TestSink()),