diff --git a/CHANGELOG.md b/CHANGELOG.md
index 615e33e..43fcf54 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,9 @@
 * Add `StreamQueue.cancelable()`, which allows users to easily make a
   `CancelableOperation` that can be canceled without affecting the queue.
 
+* Add a `subscriptionTransformer()` function to create `StreamTransformer`s that
+  modify the behavior of subscriptions to a stream.
+
 ## 1.11.3
 
 * Fix strong-mode warning against the signature of Future.then
diff --git a/lib/async.dart b/lib/async.dart
index 2526ae5..c11f0b5 100644
--- a/lib/async.dart
+++ b/lib/async.dart
@@ -29,6 +29,7 @@
 export "src/stream_sink_completer.dart";
 export "src/stream_sink_transformer.dart";
 export "src/stream_splitter.dart";
+export "src/stream_subscription_transformer.dart";
 export "src/stream_zip.dart";
 export "src/subscription_stream.dart";
 export "src/typed_stream_transformer.dart";
diff --git a/lib/src/stream_subscription_transformer.dart b/lib/src/stream_subscription_transformer.dart
new file mode 100644
index 0000000..060f466
--- /dev/null
+++ b/lib/src/stream_subscription_transformer.dart
@@ -0,0 +1,105 @@
+// 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 'async_memoizer.dart';
+import 'delegate/stream_subscription.dart';
+
+typedef Future _AsyncHandler<T>(StreamSubscription<T> inner);
+
+typedef void _VoidHandler<T>(StreamSubscription<T> inner);
+
+/// Creates a [StreamTransformer] that modifies the behavior of subscriptions to
+/// a stream.
+///
+/// When [StreamSubscription.cancel], [StreamSubscription.pause], or
+/// [StreamSubscription.resume] is called, the corresponding handler is invoked.
+/// By default, handlers just forward to the underlying subscription.
+///
+/// Guarantees that none of the [StreamSubscription] callbacks and none of the
+/// callbacks passed to `subscriptionTransformer()` will be invoked once the
+/// transformed [StreamSubscription] has been canceled and `handleCancel()` has
+/// run. The [handlePause] and [handleResume] are invoked regardless of whether
+/// the subscription is paused already or not.
+///
+/// In order to preserve [StreamSubscription] guarantees, **all callbacks must
+/// synchronously call the corresponding method** on the inner
+/// [StreamSubscription]: [handleCancel] must call `cancel()`, [handlePause]
+/// must call `pause()`, and [handleResume] must call `resume()`.
+StreamTransformer/*<T, T>*/ subscriptionTransformer/*<T>*/(
+    {Future handleCancel(StreamSubscription/*<T>*/ inner),
+    void handlePause(StreamSubscription/*<T>*/ inner),
+    void handleResume(StreamSubscription/*<T>*/ inner)}) {
+  return new StreamTransformer((stream, cancelOnError) {
+    return new _TransformedSubscription(
+        stream.listen(null, cancelOnError: cancelOnError),
+        handleCancel ?? (inner) => inner.cancel(),
+        handlePause ?? (inner) {
+          inner.pause();
+        },
+        handleResume ?? (inner) {
+          inner.resume();
+        });
+  });
+}
+
+/// A [StreamSubscription] wrapper that calls callbacks for subscription
+/// methods.
+class _TransformedSubscription<T> implements StreamSubscription<T> {
+  /// The wrapped subscription.
+  StreamSubscription<T> _inner;
+
+  /// The callback to run when [cancel] is called.
+  final _AsyncHandler<T> _handleCancel;
+
+  /// The callback to run when [pause] is called.
+  final _VoidHandler<T> _handlePause;
+
+  /// The callback to run when [resume] is called.
+  final _VoidHandler<T> _handleResume;
+
+  bool get isPaused => _inner?.isPaused ?? false;
+
+  _TransformedSubscription(this._inner, this._handleCancel, this._handlePause,
+      this._handleResume);
+
+  void onData(void handleData(T data)) {
+    _inner?.onData(handleData);
+  }
+
+  void onError(Function handleError) {
+    _inner?.onError(handleError);
+  }
+
+  void onDone(void handleDone()) {
+    _inner?.onDone(handleDone);
+  }
+
+  Future cancel() => _cancelMemoizer.runOnce(() {
+    var inner = _inner;
+    _inner.onData(null);
+    _inner.onDone(null);
+
+    // Setting onError to null will cause errors to be top-leveled.
+    _inner.onError((_, __) {});
+    _inner = null;
+    return _handleCancel(inner);
+  });
+  final _cancelMemoizer = new AsyncMemoizer();
+
+  void pause([Future resumeFuture]) {
+    if (_cancelMemoizer.hasRun) return;
+    if (resumeFuture != null) resumeFuture.whenComplete(resume);
+    _handlePause(_inner);
+  }
+
+  void resume() {
+    if (_cancelMemoizer.hasRun) return;
+    _handleResume(_inner);
+  }
+
+  Future/*<E>*/ asFuture/*<E>*/([/*=E*/ futureValue]) =>
+      _inner?.asFuture(futureValue) ?? new Completer().future;
+}
diff --git a/test/subscription_transformer_test.dart b/test/subscription_transformer_test.dart
new file mode 100644
index 0000000..cdac4f8
--- /dev/null
+++ b/test/subscription_transformer_test.dart
@@ -0,0 +1,292 @@
+// 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("with no callbacks", () {
+    test("forwards cancellation", () async {
+      var isCanceled = false;
+      var cancelCompleter = new Completer();
+      var controller = new StreamController(onCancel: expectAsync0(() {
+        isCanceled = true;
+        return cancelCompleter.future;
+      }));
+      var subscription = controller.stream
+          .transform(subscriptionTransformer())
+          .listen(expectAsync1((_) {}, count: 0));
+
+      var cancelFired = false;
+      subscription.cancel().then(expectAsync1((_) {
+        cancelFired = true;
+      }));
+
+      await flushMicrotasks();
+      expect(isCanceled, isTrue);
+      expect(cancelFired, isFalse);
+
+      cancelCompleter.complete();
+      await flushMicrotasks();
+      expect(cancelFired, isTrue);
+
+      // This shouldn't call the onCancel callback again.
+      expect(subscription.cancel(), completes);
+    });
+
+    test("forwards pausing and resuming", () async {
+      var controller = new StreamController();
+      var subscription = controller.stream
+          .transform(subscriptionTransformer())
+          .listen(expectAsync1((_) {}, count: 0));
+
+      subscription.pause();
+      await flushMicrotasks();
+      expect(controller.isPaused, isTrue);
+
+      subscription.pause();
+      await flushMicrotasks();
+      expect(controller.isPaused, isTrue);
+
+      subscription.resume();
+      await flushMicrotasks();
+      expect(controller.isPaused, isTrue);
+
+      subscription.resume();
+      await flushMicrotasks();
+      expect(controller.isPaused, isFalse);
+    });
+
+    test("forwards pausing with a resume future", () async {
+      var controller = new StreamController();
+      var subscription = controller.stream
+          .transform(subscriptionTransformer())
+          .listen(expectAsync1((_) {}, count: 0));
+
+      var completer = new Completer();
+      subscription.pause(completer.future);
+      await flushMicrotasks();
+      expect(controller.isPaused, isTrue);
+
+      completer.complete();
+      await flushMicrotasks();
+      expect(controller.isPaused, isFalse);
+    });
+  });
+
+  group("with a cancel callback", () {
+    test("invokes the callback when the subscription is canceled", () async {
+      var isCanceled = false;
+      var callbackInvoked = false;
+      var controller = new StreamController(onCancel: expectAsync0(() {
+        isCanceled = true;
+      }));
+      var subscription = controller.stream
+          .transform(subscriptionTransformer(
+              handleCancel: expectAsync1((inner) {
+                callbackInvoked = true;
+                inner.cancel();
+              })))
+          .listen(expectAsync1((_) {}, count: 0));
+
+      await flushMicrotasks();
+      expect(callbackInvoked, isFalse);
+      expect(isCanceled, isFalse);
+
+      subscription.cancel();
+      await flushMicrotasks();
+      expect(callbackInvoked, isTrue);
+      expect(isCanceled, isTrue);
+    });
+    
+    test("invokes the callback once and caches its result", () async {
+      var completer = new Completer();
+      var controller = new StreamController();
+      var subscription = controller.stream
+          .transform(subscriptionTransformer(
+              handleCancel: expectAsync1((inner) => completer.future)))
+          .listen(expectAsync1((_) {}, count: 0));
+
+      var cancelFired1 = false;
+      subscription.cancel().then(expectAsync1((_) {
+        cancelFired1 = true;
+      }));
+
+      var cancelFired2 = false;
+      subscription.cancel().then(expectAsync1((_) {
+        cancelFired2 = true;
+      }));
+
+      await flushMicrotasks();
+      expect(cancelFired1, isFalse);
+      expect(cancelFired2, isFalse);
+
+      completer.complete();
+      await flushMicrotasks();
+      expect(cancelFired1, isTrue);
+      expect(cancelFired2, isTrue);
+    });
+  });
+
+  group("with a pause callback", () {
+    test("invokes the callback when pause is called", () async {
+      var pauseCount = 0;
+      var controller = new StreamController();
+      var subscription = controller.stream
+          .transform(subscriptionTransformer(handlePause: expectAsync1((inner) {
+            pauseCount++;
+            inner.pause();
+          }, count: 3)))
+          .listen(expectAsync1((_) {}, count: 0));
+
+      await flushMicrotasks();
+      expect(pauseCount, equals(0));
+
+      subscription.pause();
+      await flushMicrotasks();
+      expect(pauseCount, equals(1));
+
+      subscription.pause();
+      await flushMicrotasks();
+      expect(pauseCount, equals(2));
+
+      subscription.resume();
+      subscription.resume();
+      await flushMicrotasks();
+      expect(pauseCount, equals(2));
+
+      subscription.pause();
+      await flushMicrotasks();
+      expect(pauseCount, equals(3));
+    });
+
+    test("doesn't invoke the callback when the subscription has been canceled",
+        () async {
+      var controller = new StreamController();
+      var subscription = controller.stream
+          .transform(subscriptionTransformer(
+              handlePause: expectAsync1((_) {}, count: 0)))
+          .listen(expectAsync1((_) {}, count: 0));
+
+      subscription.cancel();
+      subscription.pause();
+      subscription.pause();
+      subscription.pause();
+    });
+  });
+
+  group("with a resume callback", () {
+    test("invokes the callback when resume is called", () async {
+      var resumeCount = 0;
+      var controller = new StreamController();
+      var subscription = controller.stream
+          .transform(subscriptionTransformer(
+              handleResume: expectAsync1((inner) {
+                resumeCount++;
+                inner.resume();
+              }, count: 3)))
+          .listen(expectAsync1((_) {}, count: 0));
+
+      await flushMicrotasks();
+      expect(resumeCount, equals(0));
+
+      subscription.resume();
+      await flushMicrotasks();
+      expect(resumeCount, equals(1));
+
+      subscription.pause();
+      subscription.pause();
+      await flushMicrotasks();
+      expect(resumeCount, equals(1));
+
+      subscription.resume();
+      await flushMicrotasks();
+      expect(resumeCount, equals(2));
+
+      subscription.resume();
+      await flushMicrotasks();
+      expect(resumeCount, equals(3));
+    });
+
+    test("invokes the callback when a resume future completes", () async {
+      var resumed = false;
+      var controller = new StreamController();
+      var subscription = controller.stream
+          .transform(subscriptionTransformer(
+              handleResume: expectAsync1((inner) {
+                resumed = true;
+                inner.resume();
+              })))
+          .listen(expectAsync1((_) {}, count: 0));
+
+      var completer = new Completer();
+      subscription.pause(completer.future);
+      await flushMicrotasks();
+      expect(resumed, isFalse);
+
+      completer.complete();
+      await flushMicrotasks();
+      expect(resumed, isTrue);
+    });
+
+    test("doesn't invoke the callback when the subscription has been canceled",
+        () async {
+      var controller = new StreamController();
+      var subscription = controller.stream
+          .transform(subscriptionTransformer(
+              handlePause: expectAsync1((_) {}, count: 0)))
+          .listen(expectAsync1((_) {}, count: 0));
+
+      subscription.cancel();
+      subscription.resume();
+      subscription.resume();
+      subscription.resume();
+    });
+  });
+
+  group("when the outer subscription is canceled but the inner is not", () {
+    StreamSubscription subscription;
+    setUp(() {
+      var controller = new StreamController();
+      subscription = controller.stream
+          .transform(subscriptionTransformer(handleCancel: (_) {}))
+          .listen(
+              expectAsync1((_) {}, count: 0),
+              onError: expectAsync2((_, __) {}, count: 0),
+              onDone: expectAsync0(() {}, count: 0));
+      subscription.cancel();
+      controller.add(1);
+      controller.addError("oh no!");
+      controller.close();
+    });
+
+    test("doesn't call a new onData", () async {
+      subscription.onData(expectAsync1((_) {}, count: 0));
+      await flushMicrotasks();
+    });
+
+    test("doesn't call a new onError", () async {
+      subscription.onError(expectAsync2((_, __) {}, count: 0));
+      await flushMicrotasks();
+    });
+
+    test("doesn't call a new onDone", () async {
+      subscription.onDone(expectAsync0(() {}, count: 0));
+      await flushMicrotasks();
+    });
+
+    test("isPaused returns false", () {
+      expect(subscription.isPaused, isFalse);
+    });
+
+    test("asFuture never completes", () async {
+      subscription.asFuture().then(expectAsync1((_) {}, count: 0));
+      await flushMicrotasks();
+    });
+  });
+}