diff --git a/pkgs/stream_transform/.travis.yml b/pkgs/stream_transform/.travis.yml
index 24eb0d7..3b2eb08 100644
--- a/pkgs/stream_transform/.travis.yml
+++ b/pkgs/stream_transform/.travis.yml
@@ -3,7 +3,6 @@
   only: [master]
 dart:
   - dev
-  - 2.7.0
 cache:
   directories:
     - $HOME/.pub-cache
@@ -18,6 +17,3 @@
     - dart: dev
       dart_task:
         dartanalyzer: --fatal-warnings --fatal-infos .
-    - dart: 2.7.0
-      dart_task:
-        dartanalyzer: --fatal-warnings .
diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md
index 268bf35..68f8859 100644
--- a/pkgs/stream_transform/CHANGELOG.md
+++ b/pkgs/stream_transform/CHANGELOG.md
@@ -1,5 +1,6 @@
-## 1.2.1-dev
+## 2.0.0-nullsafety.0-dev
 
+- Migrate to null safety.
 - Improve tests of `switchMap` and improve documentation with links and
   clarification.
 
diff --git a/pkgs/stream_transform/example/main.dart b/pkgs/stream_transform/example/main.dart
index 4ed53d8..70b3e7f 100644
--- a/pkgs/stream_transform/example/main.dart
+++ b/pkgs/stream_transform/example/main.dart
@@ -9,7 +9,7 @@
 void main() {
   var firstInput = document.querySelector('#first_input') as InputElement;
   var secondInput = document.querySelector('#second_input') as InputElement;
-  var output = document.querySelector('#output');
+  var output = document.querySelector('#output')!;
 
   _inputValues(firstInput)
       .combineLatest(_inputValues(secondInput),
@@ -21,6 +21,6 @@
   });
 }
 
-Stream<String> _inputValues(InputElement element) => element.onKeyUp
+Stream<String?> _inputValues(InputElement element) => element.onKeyUp
     .debounce(const Duration(milliseconds: 100))
     .map((_) => element.value);
diff --git a/pkgs/stream_transform/lib/src/aggregate_sample.dart b/pkgs/stream_transform/lib/src/aggregate_sample.dart
index 88fc490..3a967d4 100644
--- a/pkgs/stream_transform/lib/src/aggregate_sample.dart
+++ b/pkgs/stream_transform/lib/src/aggregate_sample.dart
@@ -15,7 +15,7 @@
 /// the output.
 class AggregateSample<S, T> extends StreamTransformerBase<S, T> {
   final Stream<void> _trigger;
-  final T Function(S, T) _aggregate;
+  final T Function(S, T?) _aggregate;
 
   AggregateSample(this._trigger, this._aggregate);
 
@@ -25,15 +25,15 @@
         ? StreamController<T>.broadcast(sync: true)
         : StreamController<T>(sync: true);
 
-    T currentResults;
+    T? currentResults;
     var waitingForTrigger = true;
     var isTriggerDone = false;
     var isValueDone = false;
-    StreamSubscription<S> valueSub;
-    StreamSubscription<void> triggerSub;
+    StreamSubscription<S>? valueSub;
+    StreamSubscription<void>? triggerSub;
 
     void emit() {
-      controller.add(currentResults);
+      controller.add(currentResults!);
       currentResults = null;
       waitingForTrigger = true;
     }
@@ -44,7 +44,7 @@
       if (!waitingForTrigger) emit();
 
       if (isTriggerDone) {
-        valueSub.cancel();
+        valueSub!.cancel();
         controller.close();
       }
     }
@@ -63,7 +63,7 @@
       if (currentResults != null) emit();
 
       if (isValueDone) {
-        triggerSub.cancel();
+        triggerSub!.cancel();
         controller.close();
       }
     }
@@ -80,8 +80,9 @@
       assert(valueSub == null);
       valueSub = values.listen(onValue,
           onError: controller.addError, onDone: onValuesDone);
-      if (triggerSub != null) {
-        if (triggerSub.isPaused) triggerSub.resume();
+      final priorTriggerSub = triggerSub;
+      if (priorTriggerSub != null) {
+        if (priorTriggerSub.isPaused) priorTriggerSub.resume();
       } else {
         triggerSub = _trigger.listen(onTrigger,
             onError: controller.addError, onDone: onTriggerDone);
@@ -98,17 +99,16 @@
           };
       }
       controller.onCancel = () {
-        var toCancel = <StreamSubscription<void>>[];
-        if (!isValueDone) toCancel.add(valueSub);
+        var cancels = <Future<void>>[if (!isValueDone) valueSub!.cancel()];
         valueSub = null;
         if (_trigger.isBroadcast || !values.isBroadcast) {
-          if (!isTriggerDone) toCancel.add(triggerSub);
+          if (!isTriggerDone) cancels.add(triggerSub!.cancel());
           triggerSub = null;
         } else {
-          triggerSub.pause();
+          triggerSub!.pause();
         }
-        var cancels =
-            toCancel.map((s) => s.cancel()).where((f) => f != null).toList();
+        // Handle opt-out nulls
+        cancels.removeWhere((Object? f) => f == null);
         if (cancels.isEmpty) return null;
         return Future.wait(cancels).then((_) => null);
       };
diff --git a/pkgs/stream_transform/lib/src/async_map.dart b/pkgs/stream_transform/lib/src/async_map.dart
index c5b160b..7b406c7 100644
--- a/pkgs/stream_transform/lib/src/async_map.dart
+++ b/pkgs/stream_transform/lib/src/async_map.dart
@@ -117,13 +117,13 @@
 /// work.
 StreamTransformer<S, T> _asyncMapThen<S, T>(
     Future<T> Function(S) convert, void Function(void) then) {
-  Future<void> pendingEvent;
+  Future<void>? pendingEvent;
   return fromHandlers(handleData: (event, sink) {
     pendingEvent =
         convert(event).then(sink.add).catchError(sink.addError).then(then);
   }, handleDone: (sink) {
     if (pendingEvent != null) {
-      pendingEvent.then((_) => sink.close());
+      pendingEvent!.then((_) => sink.close());
     } else {
       sink.close();
     }
diff --git a/pkgs/stream_transform/lib/src/combine_latest.dart b/pkgs/stream_transform/lib/src/combine_latest.dart
index 303b16a..6e268e3 100644
--- a/pkgs/stream_transform/lib/src/combine_latest.dart
+++ b/pkgs/stream_transform/lib/src/combine_latest.dart
@@ -92,14 +92,14 @@
         ? _other.asBroadcastStream()
         : _other;
 
-    StreamSubscription<S> sourceSubscription;
-    StreamSubscription<T> otherSubscription;
+    StreamSubscription<S>? sourceSubscription;
+    StreamSubscription<T>? otherSubscription;
 
     var sourceDone = false;
     var otherDone = false;
 
-    S latestSource;
-    T latestOther;
+    late S latestSource;
+    late T latestOther;
 
     var sourceStarted = false;
     var otherStarted = false;
@@ -114,16 +114,16 @@
         return;
       }
       if (result is Future<R>) {
-        sourceSubscription.pause();
-        otherSubscription.pause();
+        sourceSubscription!.pause();
+        otherSubscription!.pause();
         result
             .then(controller.add, onError: controller.addError)
             .whenComplete(() {
-          sourceSubscription.resume();
-          otherSubscription.resume();
+          sourceSubscription!.resume();
+          otherSubscription!.resume();
         });
       } else {
-        controller.add(result as R);
+        controller.add(result);
       }
     }
 
@@ -142,7 +142,7 @@
               controller.close();
             } else if (!sourceStarted) {
               // Nothing can ever be emitted
-              otherSubscription.cancel();
+              otherSubscription!.cancel();
               controller.close();
             }
           });
@@ -159,24 +159,28 @@
               controller.close();
             } else if (!otherStarted) {
               // Nothing can ever be emitted
-              sourceSubscription.cancel();
+              sourceSubscription!.cancel();
               controller.close();
             }
           });
       if (!source.isBroadcast) {
         controller
           ..onPause = () {
-            sourceSubscription.pause();
-            otherSubscription.pause();
+            sourceSubscription!.pause();
+            otherSubscription!.pause();
           }
           ..onResume = () {
-            sourceSubscription.resume();
-            otherSubscription.resume();
+            sourceSubscription!.resume();
+            otherSubscription!.resume();
           };
       }
       controller.onCancel = () {
-        var cancels = [sourceSubscription.cancel(), otherSubscription.cancel()]
-            .where((f) => f != null);
+        var cancels = [
+          sourceSubscription!.cancel(),
+          otherSubscription!.cancel()
+        ]
+          // Handle opt-out nulls
+          ..removeWhere((Object? f) => f == null);
         sourceSubscription = null;
         otherSubscription = null;
         return Future.wait(cancels).then((_) => null);
@@ -208,7 +212,7 @@
     controller.onListen = () {
       final subscriptions = <StreamSubscription<T>>[];
 
-      final latestData = List<T>(allStreams.length);
+      final latestData = List<T?>.filled(allStreams.length, null);
       final hasEmitted = <int>{};
       void handleData(int index, T data) {
         latestData[index] = data;
@@ -249,10 +253,10 @@
           };
       }
       controller.onCancel = () {
-        var cancels = subscriptions
-            .map((s) => s.cancel())
-            .where((f) => f != null)
-            .toList();
+        if (subscriptions.isEmpty) return null;
+        var cancels = [for (var s in subscriptions) s.cancel()]
+          // Handle opt-out nulls
+          ..removeWhere((Object? f) => f == null);
         if (cancels.isEmpty) return null;
         return Future.wait(cancels).then((_) => null);
       };
diff --git a/pkgs/stream_transform/lib/src/concatenate.dart b/pkgs/stream_transform/lib/src/concatenate.dart
index 05c977f..402d8a0 100644
--- a/pkgs/stream_transform/lib/src/concatenate.dart
+++ b/pkgs/stream_transform/lib/src/concatenate.dart
@@ -68,12 +68,12 @@
         ? _next.asBroadcastStream()
         : _next;
 
-    StreamSubscription<T> subscription;
+    StreamSubscription<T>? subscription;
     var currentStream = first;
     var firstDone = false;
     var secondDone = false;
 
-    Function currentDoneHandler;
+    late void Function() currentDoneHandler;
 
     void listen() {
       subscription = currentStream.listen(controller.add,
@@ -100,18 +100,18 @@
       if (!first.isBroadcast) {
         controller
           ..onPause = () {
-            if (!firstDone || !next.isBroadcast) return subscription.pause();
-            subscription.cancel();
+            if (!firstDone || !next.isBroadcast) return subscription!.pause();
+            subscription!.cancel();
             subscription = null;
           }
           ..onResume = () {
-            if (!firstDone || !next.isBroadcast) return subscription.resume();
+            if (!firstDone || !next.isBroadcast) return subscription!.resume();
             listen();
           };
       }
       controller.onCancel = () {
         if (secondDone) return null;
-        var toCancel = subscription;
+        var toCancel = subscription!;
         subscription = null;
         return toCancel.cancel();
       };
diff --git a/pkgs/stream_transform/lib/src/from_handlers.dart b/pkgs/stream_transform/lib/src/from_handlers.dart
index 3e5689f..c7c9332 100644
--- a/pkgs/stream_transform/lib/src/from_handlers.dart
+++ b/pkgs/stream_transform/lib/src/from_handlers.dart
@@ -7,9 +7,9 @@
 /// Like [new StreamTransformer.fromHandlers] but the handlers are called once
 /// per event rather than once per listener for broadcast streams.
 StreamTransformer<S, T> fromHandlers<S, T>(
-        {void Function(S, EventSink<T>) handleData,
-        void Function(Object, StackTrace, EventSink<T>) handleError,
-        void Function(EventSink<T>) handleDone}) =>
+        {void Function(S, EventSink<T>)? handleData,
+        void Function(Object, StackTrace, EventSink<T>)? handleError,
+        void Function(EventSink<T>)? handleDone}) =>
     _StreamTransformer(
         handleData: handleData,
         handleError: handleError,
@@ -21,9 +21,9 @@
   final void Function(Object, StackTrace, EventSink<T>) _handleError;
 
   _StreamTransformer(
-      {void Function(S, EventSink<T>) handleData,
-      void Function(Object, StackTrace, EventSink<T>) handleError,
-      void Function(EventSink<T>) handleDone})
+      {void Function(S, EventSink<T>)? handleData,
+      void Function(Object, StackTrace, EventSink<T>)? handleError,
+      void Function(EventSink<T>)? handleDone})
       : _handleData = handleData ?? _defaultHandleData,
         _handleError = handleError ?? _defaultHandleError,
         _handleDone = handleDone ?? _defaultHandleDone;
@@ -47,12 +47,12 @@
         ? StreamController<T>.broadcast(sync: true)
         : StreamController<T>(sync: true);
 
-    StreamSubscription<S> subscription;
+    StreamSubscription<S>? subscription;
     controller.onListen = () {
       assert(subscription == null);
       var valuesDone = false;
       subscription = values.listen((value) => _handleData(value, controller),
-          onError: (error, StackTrace stackTrace) {
+          onError: (Object error, StackTrace stackTrace) {
         _handleError(error, stackTrace, controller);
       }, onDone: () {
         valuesDone = true;
@@ -60,13 +60,13 @@
       });
       if (!values.isBroadcast) {
         controller
-          ..onPause = subscription.pause
-          ..onResume = subscription.resume;
+          ..onPause = subscription!.pause
+          ..onResume = subscription!.resume;
       }
       controller.onCancel = () {
         var toCancel = subscription;
         subscription = null;
-        if (!valuesDone) return toCancel.cancel();
+        if (!valuesDone) return toCancel!.cancel();
         return null;
       };
     };
diff --git a/pkgs/stream_transform/lib/src/merge.dart b/pkgs/stream_transform/lib/src/merge.dart
index c68695c..61b5e21 100644
--- a/pkgs/stream_transform/lib/src/merge.dart
+++ b/pkgs/stream_transform/lib/src/merge.dart
@@ -131,10 +131,10 @@
           };
       }
       controller.onCancel = () {
-        var cancels = subscriptions
-            .map((s) => s.cancel())
-            .where((f) => f != null)
-            .toList();
+        if (subscriptions.isEmpty) return null;
+        var cancels = [for (var s in subscriptions) s.cancel()]
+          // Handle opt-out nulls
+          ..removeWhere((Object? f) => f == null);
         if (cancels.isEmpty) return null;
         return Future.wait(cancels).then((_) => null);
       };
@@ -183,11 +183,10 @@
           };
       }
       controller.onCancel = () {
-        var cancels = subscriptions
-            .map((s) => s.cancel())
-            .where((f) => f != null)
-            .toList();
-        if (cancels.isEmpty) return null;
+        if (subscriptions.isEmpty) return null;
+        var cancels = [for (var s in subscriptions) s.cancel()]
+          // Handle opt-out nulls
+          ..removeWhere((Object? f) => f == null);
         return Future.wait(cancels).then((_) => null);
       };
     };
diff --git a/pkgs/stream_transform/lib/src/rate_limit.dart b/pkgs/stream_transform/lib/src/rate_limit.dart
index e42f8f4..23dfbc5 100644
--- a/pkgs/stream_transform/lib/src/rate_limit.dart
+++ b/pkgs/stream_transform/lib/src/rate_limit.dart
@@ -85,7 +85,7 @@
   /// Events emitted by the source stream within [duration] following an emitted
   /// event will be discarded. Errors are always forwarded immediately.
   Stream<T> throttle(Duration duration) {
-    Timer timer;
+    Timer? timer;
 
     return transform(fromHandlers(handleData: (data, sink) {
       if (timer == null) {
@@ -125,7 +125,7 @@
   ///     source: a------b--c----d--|
   ///     output: -----a------c--------d|
   Stream<T> audit(Duration duration) {
-    Timer timer;
+    Timer? timer;
     var shouldClose = false;
     T recentData;
 
@@ -161,7 +161,7 @@
       transform(AggregateSample<T, List<T>>(trigger, _collect));
 }
 
-List<T> _collectToList<T>(T element, List<T> soFar) {
+List<T> _collectToList<T>(T element, List<T>? soFar) {
   soFar ??= <T>[];
   soFar.add(element);
   return soFar;
@@ -172,10 +172,10 @@
 /// Creates a StreamTransformer which aggregates values until the source stream
 /// does not emit for [duration], then emits the aggregated values.
 StreamTransformer<T, R> _debounceAggregate<T, R>(
-    Duration duration, R Function(T element, R soFar) collect,
-    {bool leading, bool trailing}) {
-  Timer timer;
-  R soFar;
+    Duration duration, R Function(T element, R? soFar) collect,
+    {required bool leading, required bool trailing}) {
+  Timer? timer;
+  R? soFar;
   var shouldClose = false;
   var emittedLatestAsLeading = false;
   return fromHandlers(handleData: (T value, EventSink<R> sink) {
@@ -183,12 +183,12 @@
     soFar = collect(value, soFar);
     if (timer == null && leading) {
       emittedLatestAsLeading = true;
-      sink.add(soFar);
+      sink.add(soFar as R);
     } else {
       emittedLatestAsLeading = false;
     }
     timer = Timer(duration, () {
-      if (trailing && !emittedLatestAsLeading) sink.add(soFar);
+      if (trailing && !emittedLatestAsLeading) sink.add(soFar as R);
       if (shouldClose) {
         sink.close();
       }
@@ -205,4 +205,4 @@
   });
 }
 
-List<T> _collect<T>(T event, List<T> soFar) => (soFar ?? <T>[])..add(event);
+List<T> _collect<T>(T event, List<T>? soFar) => (soFar ?? <T>[])..add(event);
diff --git a/pkgs/stream_transform/lib/src/scan.dart b/pkgs/stream_transform/lib/src/scan.dart
index 4e022f5..d14381c 100644
--- a/pkgs/stream_transform/lib/src/scan.dart
+++ b/pkgs/stream_transform/lib/src/scan.dart
@@ -21,7 +21,7 @@
       if (result is Future<S>) {
         return result.then((r) => accumulated = r);
       } else {
-        return accumulated = result as S;
+        return accumulated = result;
       }
     });
   }
diff --git a/pkgs/stream_transform/lib/src/switch.dart b/pkgs/stream_transform/lib/src/switch.dart
index dcd2431..df7f1b8 100644
--- a/pkgs/stream_transform/lib/src/switch.dart
+++ b/pkgs/stream_transform/lib/src/switch.dart
@@ -65,7 +65,7 @@
         : StreamController<T>(sync: true);
 
     controller.onListen = () {
-      StreamSubscription<T> innerSubscription;
+      StreamSubscription<T>? innerSubscription;
       var outerStreamDone = false;
 
       final outerSubscription = outer.listen(
@@ -96,8 +96,10 @@
       controller.onCancel = () {
         var cancels = [
           if (!outerStreamDone) outerSubscription.cancel(),
-          if (innerSubscription != null) innerSubscription.cancel(),
-        ].where((f) => f != null);
+          if (innerSubscription != null) innerSubscription!.cancel(),
+        ]
+          // Handle opt-out nulls
+          ..removeWhere((Object? f) => f == null);
         if (cancels.isEmpty) return null;
         return Future.wait(cancels).then((_) => null);
       };
diff --git a/pkgs/stream_transform/lib/src/take_until.dart b/pkgs/stream_transform/lib/src/take_until.dart
index 43b35d1..5420500 100644
--- a/pkgs/stream_transform/lib/src/take_until.dart
+++ b/pkgs/stream_transform/lib/src/take_until.dart
@@ -27,7 +27,7 @@
         ? StreamController<T>.broadcast(sync: true)
         : StreamController<T>(sync: true);
 
-    StreamSubscription<T> subscription;
+    StreamSubscription<T>? subscription;
     var isDone = false;
     _trigger.then((_) {
       if (isDone) return;
@@ -46,12 +46,12 @@
       });
       if (!values.isBroadcast) {
         controller
-          ..onPause = subscription.pause
-          ..onResume = subscription.resume;
+          ..onPause = subscription!.pause
+          ..onResume = subscription!.resume;
       }
       controller.onCancel = () {
         if (isDone) return null;
-        var toCancel = subscription;
+        var toCancel = subscription!;
         subscription = null;
         return toCancel.cancel();
       };
diff --git a/pkgs/stream_transform/lib/src/tap.dart b/pkgs/stream_transform/lib/src/tap.dart
index 2696e02..b7e0321 100644
--- a/pkgs/stream_transform/lib/src/tap.dart
+++ b/pkgs/stream_transform/lib/src/tap.dart
@@ -22,9 +22,9 @@
   ///
   /// The callbacks may not be called until the tapped stream has a listener,
   /// and may not be called after the listener has canceled the subscription.
-  Stream<T> tap(void Function(T) onValue,
-          {void Function(Object, StackTrace) onError,
-          void Function() onDone}) =>
+  Stream<T> tap(void Function(T)? onValue,
+          {void Function(Object, StackTrace)? onError,
+          void Function()? onDone}) =>
       transform(fromHandlers(handleData: (value, sink) {
         try {
           onValue?.call(value);
diff --git a/pkgs/stream_transform/pubspec.yaml b/pkgs/stream_transform/pubspec.yaml
index 5da90b1..ba4bd51 100644
--- a/pkgs/stream_transform/pubspec.yaml
+++ b/pkgs/stream_transform/pubspec.yaml
@@ -1,12 +1,12 @@
 name: stream_transform
 description: A collection of utilities to transform and manipulate streams.
 homepage: https://github.com/dart-lang/stream_transform
-version: 1.2.1-dev
+version: 2.0.0-nullsafety.0-dev
 
 environment:
-  sdk: ">=2.7.0 <3.0.0"
+  sdk: ">=2.12.0-0 <3.0.0"
 
 dev_dependencies:
-  async: ^2.0.0
-  pedantic: ^1.5.0
-  test: ^1.0.0
+  async: ^2.5.0-nullsafety
+  pedantic: ^1.10.0-nullsafety
+  test: ^1.16.0-nullsafety
diff --git a/pkgs/stream_transform/test/async_map_buffer_test.dart b/pkgs/stream_transform/test/async_map_buffer_test.dart
index 9b46e1f..1280b51 100644
--- a/pkgs/stream_transform/test/async_map_buffer_test.dart
+++ b/pkgs/stream_transform/test/async_map_buffer_test.dart
@@ -11,16 +11,16 @@
 import 'utils.dart';
 
 void main() {
-  StreamController<int> values;
-  List<String> emittedValues;
-  bool valuesCanceled;
-  bool isDone;
-  List<String> errors;
-  Stream<String> transformed;
-  StreamSubscription<String> subscription;
+  late StreamController<int> values;
+  late List<String> emittedValues;
+  late bool valuesCanceled;
+  late bool isDone;
+  late List<String> errors;
+  late Stream<String> transformed;
+  late StreamSubscription<String> subscription;
 
-  Completer<String> finishWork;
-  List<int> workArgument;
+  Completer<String>? finishWork;
+  List<int>? workArgument;
 
   /// Represents the async `convert` function and asserts that is is only called
   /// after the previous iteration has completed.
@@ -28,15 +28,15 @@
     expect(finishWork, isNull,
         reason: 'See $values befor previous work is complete');
     workArgument = values;
-    finishWork = Completer();
-    finishWork.future.then((_) {
-      workArgument = null;
-      finishWork = null;
-    }).catchError((_) {
-      workArgument = null;
-      finishWork = null;
-    });
-    return finishWork.future;
+    finishWork = Completer()
+      ..future.then((_) {
+        workArgument = null;
+        finishWork = null;
+      }).catchError((_) {
+        workArgument = null;
+        finishWork = null;
+      });
+    return finishWork!.future;
   }
 
   for (var streamType in streamTypes) {
@@ -64,7 +64,7 @@
         await Future(() {});
         expect(emittedValues, isEmpty);
         expect(workArgument, [1]);
-        finishWork.complete('result');
+        finishWork!.complete('result');
         await Future(() {});
         expect(emittedValues, ['result']);
       });
@@ -74,7 +74,7 @@
         await Future(() {});
         values..add(2)..add(3);
         await Future(() {});
-        finishWork.complete();
+        finishWork!.complete('');
         await Future(() {});
         expect(workArgument, [2, 3]);
       });
@@ -90,7 +90,7 @@
       test('forwards errors which occur during the work', () async {
         values.add(1);
         await Future(() {});
-        finishWork.completeError('error');
+        finishWork!.completeError('error');
         await Future(() {});
         expect(errors, ['error']);
       });
@@ -98,11 +98,11 @@
       test('can continue handling events after an error', () async {
         values.add(1);
         await Future(() {});
-        finishWork.completeError('error');
+        finishWork!.completeError('error');
         values.add(2);
         await Future(() {});
         expect(workArgument, [2]);
-        finishWork.completeError('another');
+        finishWork!.completeError('another');
         await Future(() {});
         expect(errors, ['error', 'another']);
       });
@@ -140,11 +140,11 @@
         values.add(2);
         await values.close();
         expect(isDone, false);
-        finishWork.complete(null);
+        finishWork!.complete('');
         await Future(() {});
         // Still a pending value
         expect(isDone, false);
-        finishWork.complete(null);
+        finishWork!.complete('');
         await Future(() {});
         expect(isDone, true);
       });
@@ -161,7 +161,7 @@
           transformed.listen(otherValues.add);
           values.add(1);
           await Future(() {});
-          finishWork.complete('result');
+          finishWork!.complete('result');
           await Future(() {});
           expect(emittedValues, ['result']);
           expect(otherValues, ['result']);
@@ -175,7 +175,7 @@
           await values.close();
           expect(isDone, false);
           expect(otherDone, false);
-          finishWork.complete();
+          finishWork!.complete('');
           await Future(() {});
           expect(isDone, true);
           expect(otherDone, true);
@@ -184,7 +184,7 @@
         test('can cancel and relisten', () async {
           values.add(1);
           await Future(() {});
-          finishWork.complete('first');
+          finishWork!.complete('first');
           await Future(() {});
           await subscription.cancel();
           values.add(2);
@@ -193,7 +193,7 @@
           values.add(3);
           await Future(() {});
           expect(workArgument, [3]);
-          finishWork.complete('second');
+          finishWork!.complete('second');
           await Future(() {});
           expect(emittedValues, ['first', 'second']);
         });
diff --git a/pkgs/stream_transform/test/async_map_sample_test.dart b/pkgs/stream_transform/test/async_map_sample_test.dart
index 9c37cd4..06457d8 100644
--- a/pkgs/stream_transform/test/async_map_sample_test.dart
+++ b/pkgs/stream_transform/test/async_map_sample_test.dart
@@ -11,16 +11,16 @@
 import 'utils.dart';
 
 void main() {
-  StreamController<int> values;
-  List<String> emittedValues;
-  bool valuesCanceled;
-  bool isDone;
-  List<String> errors;
-  Stream<String> transformed;
-  StreamSubscription<String> subscription;
+  late StreamController<int> values;
+  late List<String> emittedValues;
+  late bool valuesCanceled;
+  late bool isDone;
+  late List<String> errors;
+  late Stream<String> transformed;
+  late StreamSubscription<String> subscription;
 
-  Completer<String> finishWork;
-  int workArgument;
+  Completer<String>? finishWork;
+  int? workArgument;
 
   /// Represents the async `convert` function and asserts that is is only called
   /// after the previous iteration has completed.
@@ -28,15 +28,15 @@
     expect(finishWork, isNull,
         reason: 'See $values befor previous work is complete');
     workArgument = value;
-    finishWork = Completer();
-    finishWork.future.then((_) {
-      workArgument = null;
-      finishWork = null;
-    }).catchError((_) {
-      workArgument = null;
-      finishWork = null;
-    });
-    return finishWork.future;
+    finishWork = Completer()
+      ..future.then((_) {
+        workArgument = null;
+        finishWork = null;
+      }).catchError((_) {
+        workArgument = null;
+        finishWork = null;
+      });
+    return finishWork!.future;
   }
 
   for (var streamType in streamTypes) {
@@ -64,7 +64,7 @@
         await Future(() {});
         expect(emittedValues, isEmpty);
         expect(workArgument, 1);
-        finishWork.complete('result');
+        finishWork!.complete('result');
         await Future(() {});
         expect(emittedValues, ['result']);
       });
@@ -74,7 +74,7 @@
         await Future(() {});
         values..add(2)..add(3);
         await Future(() {});
-        finishWork.complete();
+        finishWork!.complete('');
         await Future(() {});
         expect(workArgument, 3);
       });
@@ -90,7 +90,7 @@
       test('forwards errors which occur during the work', () async {
         values.add(1);
         await Future(() {});
-        finishWork.completeError('error');
+        finishWork!.completeError('error');
         await Future(() {});
         expect(errors, ['error']);
       });
@@ -98,11 +98,11 @@
       test('can continue handling events after an error', () async {
         values.add(1);
         await Future(() {});
-        finishWork.completeError('error');
+        finishWork!.completeError('error');
         values.add(2);
         await Future(() {});
         expect(workArgument, 2);
-        finishWork.completeError('another');
+        finishWork!.completeError('another');
         await Future(() {});
         expect(errors, ['error', 'another']);
       });
@@ -140,11 +140,11 @@
         values.add(2);
         await values.close();
         expect(isDone, false);
-        finishWork.complete(null);
+        finishWork!.complete('');
         await Future(() {});
         // Still a pending value
         expect(isDone, false);
-        finishWork.complete(null);
+        finishWork!.complete('');
         await Future(() {});
         expect(isDone, true);
       });
@@ -161,7 +161,7 @@
           transformed.listen(otherValues.add);
           values.add(1);
           await Future(() {});
-          finishWork.complete('result');
+          finishWork!.complete('result');
           await Future(() {});
           expect(emittedValues, ['result']);
           expect(otherValues, ['result']);
@@ -175,7 +175,7 @@
           await values.close();
           expect(isDone, false);
           expect(otherDone, false);
-          finishWork.complete();
+          finishWork!.complete('');
           await Future(() {});
           expect(isDone, true);
           expect(otherDone, true);
@@ -184,7 +184,7 @@
         test('can cancel and relisten', () async {
           values.add(1);
           await Future(() {});
-          finishWork.complete('first');
+          finishWork!.complete('first');
           await Future(() {});
           await subscription.cancel();
           values.add(2);
@@ -193,20 +193,11 @@
           values.add(3);
           await Future(() {});
           expect(workArgument, 3);
-          finishWork.complete('second');
+          finishWork!.complete('second');
           await Future(() {});
           expect(emittedValues, ['first', 'second']);
         });
       }
     });
   }
-  test('handles null response from cancel', () async {
-    var controller = StreamController<int>();
-
-    var subscription = NullOnCancelStream(controller.stream)
-        .asyncMapSample((_) async {})
-        .listen(null);
-
-    await subscription.cancel();
-  });
 }
diff --git a/pkgs/stream_transform/test/audit_test.dart b/pkgs/stream_transform/test/audit_test.dart
index 58cb95b..20c7d9f 100644
--- a/pkgs/stream_transform/test/audit_test.dart
+++ b/pkgs/stream_transform/test/audit_test.dart
@@ -12,13 +12,13 @@
 void main() {
   for (var streamType in streamTypes) {
     group('Stream type [$streamType]', () {
-      StreamController<int> values;
-      List<int> emittedValues;
-      bool valuesCanceled;
-      bool isDone;
-      List<String> errors;
-      Stream<int> transformed;
-      StreamSubscription<int> subscription;
+      late StreamController<int> values;
+      late List<int> emittedValues;
+      late bool valuesCanceled;
+      late bool isDone;
+      late List<String> errors;
+      late Stream<int> transformed;
+      late StreamSubscription<int> subscription;
 
       group('audit', () {
         setUp(() async {
diff --git a/pkgs/stream_transform/test/buffer_test.dart b/pkgs/stream_transform/test/buffer_test.dart
index 80a793c..b528fa2 100644
--- a/pkgs/stream_transform/test/buffer_test.dart
+++ b/pkgs/stream_transform/test/buffer_test.dart
@@ -10,16 +10,16 @@
 import 'utils.dart';
 
 void main() {
-  StreamController<void> trigger;
-  StreamController<int> values;
-  List<List<int>> emittedValues;
-  bool valuesCanceled;
-  bool triggerCanceled;
-  bool triggerPaused;
-  bool isDone;
-  List<String> errors;
-  Stream<List<int>> transformed;
-  StreamSubscription<List<int>> subscription;
+  late StreamController<void> trigger;
+  late StreamController<int> values;
+  late List<List<int>> emittedValues;
+  late bool valuesCanceled;
+  late bool triggerCanceled;
+  late bool triggerPaused;
+  late bool isDone;
+  late List<String> errors;
+  late Stream<List<int>> transformed;
+  late StreamSubscription<List<int>> subscription;
 
   void setUpForStreamTypes(String triggerType, String valuesType) {
     valuesCanceled = false;
@@ -243,13 +243,4 @@
       ]);
     });
   }
-
-  test('handles null response from cancel', () async {
-    var controller = StreamController<int>();
-    var trigger = StreamController<void>();
-    var subscription = NullOnCancelStream(controller.stream)
-        .buffer(trigger.stream)
-        .listen(null);
-    await subscription.cancel();
-  });
 }
diff --git a/pkgs/stream_transform/test/combine_latest_all_test.dart b/pkgs/stream_transform/test/combine_latest_all_test.dart
index d57d618..2943449 100644
--- a/pkgs/stream_transform/test/combine_latest_all_test.dart
+++ b/pkgs/stream_transform/test/combine_latest_all_test.dart
@@ -8,8 +8,6 @@
 
 import 'package:stream_transform/stream_transform.dart';
 
-import 'utils.dart';
-
 Future<void> tick() => Future(() {});
 
 void main() {
@@ -166,14 +164,4 @@
       });
     });
   });
-
-  test('handles null response from cancel', () async {
-    var source = StreamController<int>();
-    var other = StreamController<int>();
-
-    var subscription = NullOnCancelStream(source.stream)
-        .combineLatestAll([NullOnCancelStream(other.stream)]).listen(null);
-
-    await subscription.cancel();
-  });
 }
diff --git a/pkgs/stream_transform/test/combine_latest_test.dart b/pkgs/stream_transform/test/combine_latest_test.dart
index 4ea9499..2b35e2c 100644
--- a/pkgs/stream_transform/test/combine_latest_test.dart
+++ b/pkgs/stream_transform/test/combine_latest_test.dart
@@ -9,8 +9,6 @@
 
 import 'package:stream_transform/stream_transform.dart';
 
-import 'utils.dart';
-
 void main() {
   group('combineLatest', () {
     test('flows through combine callback', () async {
@@ -170,17 +168,6 @@
       });
     });
   });
-
-  test('handles null response from cancel', () async {
-    var source = StreamController<int>();
-    var other = StreamController<int>();
-
-    var subscription = NullOnCancelStream(source.stream)
-        .combineLatest(NullOnCancelStream(other.stream), (a, b) => null)
-        .listen(null);
-
-    await subscription.cancel();
-  });
 }
 
 class _NumberedException implements Exception {
diff --git a/pkgs/stream_transform/test/concurrent_async_expand_test.dart b/pkgs/stream_transform/test/concurrent_async_expand_test.dart
index 9ed02d0..d0b4ad4 100644
--- a/pkgs/stream_transform/test/concurrent_async_expand_test.dart
+++ b/pkgs/stream_transform/test/concurrent_async_expand_test.dart
@@ -24,15 +24,15 @@
   for (var outerType in streamTypes) {
     for (var innerType in streamTypes) {
       group('concurrentAsyncExpand $outerType to $innerType', () {
-        StreamController<int> outerController;
-        bool outerCanceled;
-        List<StreamController<String>> innerControllers;
-        List<bool> innerCanceled;
-        List<String> emittedValues;
-        bool isDone;
-        List<String> errors;
-        Stream<String> transformed;
-        StreamSubscription<String> subscription;
+        late StreamController<int> outerController;
+        late bool outerCanceled;
+        late List<StreamController<String>> innerControllers;
+        late List<bool> innerCanceled;
+        late List<String> emittedValues;
+        late bool isDone;
+        late List<String> errors;
+        late Stream<String> transformed;
+        late StreamSubscription<String> subscription;
 
         setUp(() {
           outerController = createController(outerType)
@@ -184,19 +184,4 @@
       });
     }
   }
-
-  test('hendles null response from cancel', () async {
-    var source = StreamController<int>();
-    var other = StreamController<int>();
-
-    var subscription = NullOnCancelStream(source.stream)
-        .concurrentAsyncExpand((_) => NullOnCancelStream(other.stream))
-        .listen(null);
-
-    source.add(1);
-
-    await Future<void>(() {});
-
-    await subscription.cancel();
-  });
 }
diff --git a/pkgs/stream_transform/test/concurrent_async_map_test.dart b/pkgs/stream_transform/test/concurrent_async_map_test.dart
index 6da2101..a894e7f 100644
--- a/pkgs/stream_transform/test/concurrent_async_map_test.dart
+++ b/pkgs/stream_transform/test/concurrent_async_map_test.dart
@@ -11,16 +11,16 @@
 import 'utils.dart';
 
 void main() {
-  StreamController<int> controller;
-  List<String> emittedValues;
-  bool valuesCanceled;
-  bool isDone;
-  List<String> errors;
-  Stream<String> transformed;
-  StreamSubscription<String> subscription;
+  late StreamController<int> controller;
+  late List<String> emittedValues;
+  late bool valuesCanceled;
+  late bool isDone;
+  late List<String> errors;
+  late Stream<String> transformed;
+  late StreamSubscription<String> subscription;
 
-  List<Completer<String>> finishWork;
-  List<dynamic> values;
+  late List<Completer<String>> finishWork;
+  late List<dynamic> values;
 
   Future<String> convert(int value) {
     values.add(value);
@@ -127,7 +127,7 @@
           await controller.close();
           expect(isDone, false);
           expect(otherDone, false);
-          finishWork.first.complete();
+          finishWork.first.complete('');
           await Future(() {});
           expect(isDone, true);
           expect(otherDone, true);
diff --git a/pkgs/stream_transform/test/debounce_test.dart b/pkgs/stream_transform/test/debounce_test.dart
index de6b740..f24bfb9 100644
--- a/pkgs/stream_transform/test/debounce_test.dart
+++ b/pkgs/stream_transform/test/debounce_test.dart
@@ -13,13 +13,13 @@
   for (var streamType in streamTypes) {
     group('Stream type [$streamType]', () {
       group('debounce - trailing', () {
-        StreamController<int> values;
-        List<int> emittedValues;
-        bool valuesCanceled;
-        bool isDone;
-        List<String> errors;
-        StreamSubscription<int> subscription;
-        Stream<int> transformed;
+        late StreamController<int> values;
+        late List<int> emittedValues;
+        late bool valuesCanceled;
+        late bool isDone;
+        late List<String> errors;
+        late StreamSubscription<int> subscription;
+        late Stream<int> transformed;
 
         setUp(() async {
           valuesCanceled = false;
@@ -89,10 +89,10 @@
       });
 
       group('debounce - leading', () {
-        StreamController<int> values;
-        List<int> emittedValues;
-        Stream<int> transformed;
-        bool isDone;
+        late StreamController<int> values;
+        late List<int> emittedValues;
+        late Stream<int> transformed;
+        late bool isDone;
 
         setUp(() async {
           values = createController(streamType);
@@ -139,9 +139,9 @@
       });
 
       group('debounce - leading and trailing', () {
-        StreamController<int> values;
-        List<int> emittedValues;
-        Stream<int> transformed;
+        late StreamController<int> values;
+        late List<int> emittedValues;
+        late Stream<int> transformed;
 
         setUp(() async {
           values = createController(streamType);
@@ -179,10 +179,10 @@
       });
 
       group('debounceBuffer', () {
-        StreamController<int> values;
-        List<List<int>> emittedValues;
-        List<String> errors;
-        Stream<List<int>> transformed;
+        late StreamController<int> values;
+        late List<List<int>> emittedValues;
+        late List<String> errors;
+        late Stream<List<int>> transformed;
 
         setUp(() async {
           values = createController(streamType);
diff --git a/pkgs/stream_transform/test/followd_by_test.dart b/pkgs/stream_transform/test/followd_by_test.dart
index c8864b5..fbe904a 100644
--- a/pkgs/stream_transform/test/followd_by_test.dart
+++ b/pkgs/stream_transform/test/followd_by_test.dart
@@ -13,17 +13,17 @@
   for (var firstType in streamTypes) {
     for (var secondType in streamTypes) {
       group('followedBy [$firstType] with [$secondType]', () {
-        StreamController<int> first;
-        StreamController<int> second;
+        late StreamController<int> first;
+        late StreamController<int> second;
 
-        List<int> emittedValues;
-        bool firstCanceled;
-        bool secondCanceled;
-        bool secondListened;
-        bool isDone;
-        List<String> errors;
-        Stream<int> transformed;
-        StreamSubscription<int> subscription;
+        late List<int> emittedValues;
+        late bool firstCanceled;
+        late bool secondCanceled;
+        late bool secondListened;
+        late bool isDone;
+        late List<String> errors;
+        late Stream<int> transformed;
+        late StreamSubscription<int> subscription;
 
         setUp(() async {
           firstCanceled = false;
diff --git a/pkgs/stream_transform/test/from_handlers_test.dart b/pkgs/stream_transform/test/from_handlers_test.dart
index 50d59c5..206acc8 100644
--- a/pkgs/stream_transform/test/from_handlers_test.dart
+++ b/pkgs/stream_transform/test/from_handlers_test.dart
@@ -9,13 +9,13 @@
 import 'package:stream_transform/src/from_handlers.dart';
 
 void main() {
-  StreamController<int> values;
-  List<int> emittedValues;
-  bool valuesCanceled;
-  bool isDone;
-  List<String> errors;
-  Stream<int> transformed;
-  StreamSubscription<int> subscription;
+  late StreamController<int> values;
+  late List<int> emittedValues;
+  late bool valuesCanceled;
+  late bool isDone;
+  late List<String> errors;
+  late Stream<int> transformed;
+  late StreamSubscription<int> subscription;
 
   void setUpForController(StreamController<int> controller,
       StreamTransformer<int, int> transformer) {
@@ -68,10 +68,10 @@
     });
 
     group('broadcast stream with muliple listeners', () {
-      List<int> emittedValues2;
-      List<String> errors2;
-      bool isDone2;
-      StreamSubscription<int> subscription2;
+      late List<int> emittedValues2;
+      late List<String> errors2;
+      late bool isDone2;
+      late StreamSubscription<int> subscription2;
 
       setUp(() {
         setUpForController(StreamController.broadcast(), fromHandlers());
@@ -133,9 +133,9 @@
     });
 
     group('broadcast stream with multiple listeners', () {
-      int dataCallCount;
-      int doneCallCount;
-      int errorCallCount;
+      late int dataCallCount;
+      late int doneCallCount;
+      late int errorCallCount;
 
       setUp(() async {
         dataCallCount = 0;
diff --git a/pkgs/stream_transform/test/merge_test.dart b/pkgs/stream_transform/test/merge_test.dart
index 2eef5d1..24cd76b 100644
--- a/pkgs/stream_transform/test/merge_test.dart
+++ b/pkgs/stream_transform/test/merge_test.dart
@@ -8,8 +8,6 @@
 
 import 'package:stream_transform/stream_transform.dart';
 
-import 'utils.dart';
-
 void main() {
   group('merge', () {
     test('includes all values', () async {
@@ -140,15 +138,4 @@
       expect(secondListenerValues, [1, 2, 3, 4, 5, 6]);
     });
   });
-
-  test('handles null response rom cancel', () async {
-    var source = StreamController<int>();
-    var other = StreamController<int>();
-
-    var subscription = NullOnCancelStream(source.stream)
-        .merge(NullOnCancelStream(other.stream))
-        .listen(null);
-
-    await subscription.cancel();
-  });
 }
diff --git a/pkgs/stream_transform/test/opt_out_test.dart b/pkgs/stream_transform/test/opt_out_test.dart
new file mode 100644
index 0000000..8ab4fc4
--- /dev/null
+++ b/pkgs/stream_transform/test/opt_out_test.dart
@@ -0,0 +1,72 @@
+// Copyright (c) 2020, 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.
+
+// @dart=2.9
+
+import 'dart:async';
+
+import 'package:async/async.dart';
+import 'package:test/test.dart';
+
+import 'package:stream_transform/stream_transform.dart';
+
+void main() {
+  group('null returned from StreamSubscription.cancel', () {
+    void testNullCancel(
+        String name, Stream<void> Function(Stream<void>) transform) {
+      test(name, () async {
+        var subscription = transform(_NullOnCancelStream()).listen(null);
+        await subscription.cancel();
+      });
+    }
+
+    testNullCancel('asyncMapSample', (s) => s.asyncMapSample((_) async {}));
+    testNullCancel('buffer', (s) => s.buffer(_nonEndingStream()));
+    testNullCancel(
+        'combineLatestAll', (s) => s.combineLatestAll([_NullOnCancelStream()]));
+    testNullCancel('combineLatest',
+        (s) => s.combineLatest(_NullOnCancelStream(), (a, b) {}));
+    testNullCancel('merge', (s) => s.merge(_NullOnCancelStream()));
+
+    test('switchLatest', () async {
+      var subscription =
+          _NullOnCancelStream(Stream<Stream<void>>.value(_NullOnCancelStream()))
+              .switchLatest()
+              .listen(null);
+      await Future(() {});
+      await subscription.cancel();
+    });
+
+    test('concurrentAsyncExpand', () async {
+      var subscription = _NullOnCancelStream(Stream.value(null))
+          .concurrentAsyncExpand((_) => _NullOnCancelStream())
+          .listen(null);
+      await Future(() {});
+      await subscription.cancel();
+    });
+  });
+}
+
+class _NullOnCancelStream<T> extends StreamView<T> {
+  _NullOnCancelStream([Stream<T> stream]) : super(stream ?? _nonEndingStream());
+
+  @override
+  StreamSubscription<T> listen(void Function(T) onData,
+          {Function onError, void Function() onDone, bool cancelOnError}) =>
+      _NullOnCancelSubscription(super.listen(onData,
+          onError: onError, onDone: onDone, cancelOnError: cancelOnError));
+}
+
+class _NullOnCancelSubscription<T> extends DelegatingStreamSubscription<T> {
+  final StreamSubscription<T> _subscription;
+  _NullOnCancelSubscription(this._subscription) : super(_subscription);
+
+  @override
+  Future<void> cancel() {
+    _subscription.cancel();
+    return null;
+  }
+}
+
+Stream<T> _nonEndingStream<T>() => StreamController<T>().stream;
diff --git a/pkgs/stream_transform/test/scan_test.dart b/pkgs/stream_transform/test/scan_test.dart
index 4e58680..e3effb3 100644
--- a/pkgs/stream_transform/test/scan_test.dart
+++ b/pkgs/stream_transform/test/scan_test.dart
@@ -22,7 +22,7 @@
     test('can create a broadcast stream', () {
       var source = StreamController.broadcast();
 
-      var transformed = source.stream.scan(null, null);
+      var transformed = source.stream.scan(null, (_, __) {});
 
       expect(transformed.isBroadcast, true);
     });
diff --git a/pkgs/stream_transform/test/start_with_test.dart b/pkgs/stream_transform/test/start_with_test.dart
index 3a5a025..29241be 100644
--- a/pkgs/stream_transform/test/start_with_test.dart
+++ b/pkgs/stream_transform/test/start_with_test.dart
@@ -11,12 +11,12 @@
 import 'utils.dart';
 
 void main() {
-  StreamController<int> values;
-  Stream<int> transformed;
-  StreamSubscription<int> subscription;
+  late StreamController<int> values;
+  late Stream<int> transformed;
+  late StreamSubscription<int> subscription;
 
-  List<int> emittedValues;
-  bool isDone;
+  late List<int> emittedValues;
+  late bool isDone;
 
   void setupForStreamType(
       String streamType, Stream<int> Function(Stream<int>) transform) {
@@ -102,7 +102,7 @@
 
     for (var startingStreamType in streamTypes) {
       group('startWithStream [$startingStreamType] then [$streamType]', () {
-        StreamController<int> starting;
+        late StreamController<int> starting;
         setUp(() async {
           starting = createController(startingStreamType);
           setupForStreamType(
diff --git a/pkgs/stream_transform/test/switch_test.dart b/pkgs/stream_transform/test/switch_test.dart
index 16deb4b..b69c8b0 100644
--- a/pkgs/stream_transform/test/switch_test.dart
+++ b/pkgs/stream_transform/test/switch_test.dart
@@ -14,16 +14,16 @@
   for (var outerType in streamTypes) {
     for (var innerType in streamTypes) {
       group('Outer type: [$outerType], Inner type: [$innerType]', () {
-        StreamController<int> first;
-        StreamController<int> second;
-        StreamController<Stream<int>> outer;
+        late StreamController<int> first;
+        late StreamController<int> second;
+        late StreamController<Stream<int>> outer;
 
-        List<int> emittedValues;
-        bool firstCanceled;
-        bool outerCanceled;
-        bool isDone;
-        List<String> errors;
-        StreamSubscription<int> subscription;
+        late List<int> emittedValues;
+        late bool firstCanceled;
+        late bool outerCanceled;
+        late bool isDone;
+        late List<String> errors;
+        late StreamSubscription<int> subscription;
 
         setUp(() async {
           firstCanceled = false;
@@ -141,24 +141,11 @@
     test('can create a broadcast stream', () async {
       var outer = StreamController.broadcast();
 
-      var transformed = outer.stream.switchMap(null);
+      var transformed = outer.stream.switchMap((_) => const Stream.empty());
 
       expect(transformed.isBroadcast, true);
     });
 
-    test('handles null response from cancel', () async {
-      var outer = StreamController<Stream<int>>();
-      var inner = StreamController<int>();
-
-      var subscription =
-          NullOnCancelStream(outer.stream).switchLatest().listen(null);
-
-      outer.add(NullOnCancelStream(inner.stream));
-      await Future<void>(() {});
-
-      await subscription.cancel();
-    });
-
     test('forwards errors from the convert callback', () async {
       var errors = <String>[];
       var source = Stream.fromIterable([1, 2, 3]);
diff --git a/pkgs/stream_transform/test/take_until_test.dart b/pkgs/stream_transform/test/take_until_test.dart
index c25d342..11e906f 100644
--- a/pkgs/stream_transform/test/take_until_test.dart
+++ b/pkgs/stream_transform/test/take_until_test.dart
@@ -13,14 +13,14 @@
 void main() {
   for (var streamType in streamTypes) {
     group('takeUntil on Stream type [$streamType]', () {
-      StreamController<int> values;
-      List<int> emittedValues;
-      bool valuesCanceled;
-      bool isDone;
-      List<String> errors;
-      Stream<int> transformed;
-      StreamSubscription<int> subscription;
-      Completer<void> closeTrigger;
+      late StreamController<int> values;
+      late List<int> emittedValues;
+      late bool valuesCanceled;
+      late bool isDone;
+      late List<String> errors;
+      late Stream<int> transformed;
+      late StreamSubscription<int> subscription;
+      late Completer<void> closeTrigger;
 
       setUp(() {
         valuesCanceled = false;
diff --git a/pkgs/stream_transform/test/throttle_test.dart b/pkgs/stream_transform/test/throttle_test.dart
index e51ef84..d84fdf4 100644
--- a/pkgs/stream_transform/test/throttle_test.dart
+++ b/pkgs/stream_transform/test/throttle_test.dart
@@ -12,12 +12,12 @@
 void main() {
   for (var streamType in streamTypes) {
     group('Stream type [$streamType]', () {
-      StreamController<int> values;
-      List<int> emittedValues;
-      bool valuesCanceled;
-      bool isDone;
-      Stream<int> transformed;
-      StreamSubscription<int> subscription;
+      late StreamController<int> values;
+      late List<int> emittedValues;
+      late bool valuesCanceled;
+      late bool isDone;
+      late Stream<int> transformed;
+      late StreamSubscription<int> subscription;
 
       group('throttle', () {
         setUp(() async {
diff --git a/pkgs/stream_transform/test/utils.dart b/pkgs/stream_transform/test/utils.dart
index 4aa5837..b6196d6 100644
--- a/pkgs/stream_transform/test/utils.dart
+++ b/pkgs/stream_transform/test/utils.dart
@@ -4,8 +4,6 @@
 
 import 'dart:async';
 
-import 'package:async/async.dart';
-
 /// Cycle the event loop to ensure timers are started, then wait for a delay
 /// longer than [milliseconds] to allow for the timer to fire.
 Future<void> waitForTimer(int milliseconds) =>
@@ -25,26 +23,3 @@
 }
 
 const streamTypes = ['single subscription', 'broadcast'];
-
-class NullOnCancelStream<T> extends StreamView<T> {
-  final Stream<T> _stream;
-
-  NullOnCancelStream(this._stream) : super(_stream);
-
-  @override
-  StreamSubscription<T> listen(void Function(T) onData,
-          {Function onError, void Function() onDone, bool cancelOnError}) =>
-      _NullOnCancelSubscription(_stream.listen(onData,
-          onError: onError, onDone: onDone, cancelOnError: cancelOnError));
-}
-
-class _NullOnCancelSubscription<T> extends DelegatingStreamSubscription<T> {
-  final StreamSubscription<T> _subscription;
-  _NullOnCancelSubscription(this._subscription) : super(_subscription);
-
-  @override
-  Future<void> cancel() {
-    _subscription.cancel();
-    return null;
-  }
-}