diff --git a/.travis.yml b/.travis.yml
index 6fb2f8d..0789302 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,18 +1,33 @@
 language: dart
 
 dart:
-  - dev
-  - 2.2.0
+ - dev
 
-dart_task:
-  - test: --platform vm
-  - test: --platform chrome
-  - dartanalyzer
-  - dartfmt
+jobs:
+  include:
+    - stage: analyze_and_format
+      name: "Analyzer"
+      dart: be/raw/latest
+      os: linux
+      script: dartanalyzer --enable-experiment=non-nullable --fatal-warnings --fatal-infos .
+    - stage: analyze_and_format
+      name: "Format"
+      dart: be/raw/latest
+      os: linux
+      script: dartfmt -n --set-exit-if-changed .
+    - stage: test
+      name: "Vm Tests"
+      dart: be/raw/latest
+      os: linux
+      script: pub run --enable-experiment=non-nullable test -p vm
+
+stages:
+  - analyze_and_format
+  - test
 
 # Only building master means that we don't run two builds for each pull request.
 branches:
-  only: [master]
+  only: [master, null_safety]
 
 cache:
   directories:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f247608..3ca7f6c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 2.5.0-nullsafety
+
+* Migrate this package to null safety.
+
 ## 2.4.2
 
 * `StreamQueue` starts listening immediately to broadcast strings.
diff --git a/analysis_options.yaml b/analysis_options.yaml
index 18c26e0..b58ef99 100644
--- a/analysis_options.yaml
+++ b/analysis_options.yaml
@@ -7,6 +7,8 @@
     todo: ignore
     # Lint provided by pkg:pedantic – should fix this!
     unawaited_futures: ignore
+  enable-experiment:
+    - non-nullable
 
 linter:
   rules:
diff --git a/lib/src/async_cache.dart b/lib/src/async_cache.dart
index e0a6f49..d6f5f7f 100644
--- a/lib/src/async_cache.dart
+++ b/lib/src/async_cache.dart
@@ -29,13 +29,13 @@
   final Duration _duration;
 
   /// Cached results of a previous [fetchStream] call.
-  StreamSplitter<T> _cachedStreamSplitter;
+  StreamSplitter<T>? _cachedStreamSplitter;
 
   /// Cached results of a previous [fetch] call.
-  Future<T> _cachedValueFuture;
+  Future<T>? _cachedValueFuture;
 
   /// Fires when the cache should be considered stale.
-  Timer _stale;
+  Timer? _stale;
 
   /// Creates a cache that invalidates its contents after [duration] has passed.
   ///
@@ -65,7 +65,7 @@
       await _cachedValueFuture;
       _startStaleTimer();
     }
-    return _cachedValueFuture;
+    return _cachedValueFuture!;
   }
 
   /// Returns a cached stream from a previous call to [fetchStream], or runs
@@ -78,12 +78,12 @@
     if (_cachedValueFuture != null) {
       throw StateError('Previously used to cache via `fetch`');
     }
-    _cachedStreamSplitter ??= StreamSplitter(
+    var splitter = _cachedStreamSplitter ??= StreamSplitter(
         callback().transform(StreamTransformer.fromHandlers(handleDone: (sink) {
       _startStaleTimer();
       sink.close();
     })));
-    return _cachedStreamSplitter.split();
+    return splitter.split();
   }
 
   /// Removes any cached value.
diff --git a/lib/src/cancelable_operation.dart b/lib/src/cancelable_operation.dart
index 9fd0534..267c6d1 100644
--- a/lib/src/cancelable_operation.dart
+++ b/lib/src/cancelable_operation.dart
@@ -35,7 +35,7 @@
   /// moment this [CancelableOperation] is created, regardless of whether
   /// [inner] has completed yet or not.
   factory CancelableOperation.fromFuture(Future<T> inner,
-      {FutureOr Function() onCancel}) {
+      {FutureOr Function()? onCancel}) {
     var completer = CancelableCompleter<T>(onCancel: onCancel);
     completer.complete(inner);
     return completer.operation;
@@ -55,7 +55,7 @@
     value.then((value) {
       controller.add(value);
       controller.close();
-    }, onError: (error, StackTrace stackTrace) {
+    }, onError: (Object error, StackTrace stackTrace) {
       controller.addError(error, stackTrace);
       controller.close();
     });
@@ -68,8 +68,8 @@
   /// If this operation completes, this completes to the same result as [value].
   /// If this operation is cancelled, the returned future waits for the future
   /// returned by [cancel], then completes to [cancellationValue].
-  Future<T> valueOrCancellation([T cancellationValue]) {
-    var completer = Completer<T>.sync();
+  Future<T?> valueOrCancellation([T? cancellationValue]) {
+    var completer = Completer<T?>.sync();
     value.then((result) => completer.complete(result),
         onError: completer.completeError);
 
@@ -93,23 +93,24 @@
   /// If [propagateCancel] is `true` and the returned operation is canceled then
   /// this operation is canceled. The default is `false`.
   CancelableOperation<R> then<R>(FutureOr<R> Function(T) onValue,
-      {FutureOr<R> Function(Object, StackTrace) onError,
-      FutureOr<R> Function() onCancel,
+      {FutureOr<R> Function(Object, StackTrace)? onError,
+      FutureOr<R> Function()? onCancel,
       bool propagateCancel = false}) {
     final completer =
         CancelableCompleter<R>(onCancel: propagateCancel ? cancel : null);
 
-    valueOrCancellation().then((T result) {
+    valueOrCancellation().then((T? result) {
       if (!completer.isCanceled) {
         if (isCompleted) {
-          completer.complete(Future.sync(() => onValue(result)));
+          assert(result is T);
+          completer.complete(Future.sync(() => onValue(result!)));
         } else if (onCancel != null) {
           completer.complete(Future.sync(onCancel));
         } else {
           completer._cancel();
         }
       }
-    }, onError: (error, StackTrace stackTrace) {
+    }, onError: (Object error, StackTrace stackTrace) {
       if (!completer.isCanceled) {
         if (onError != null) {
           completer.complete(Future.sync(() => onError(error, stackTrace)));
@@ -145,7 +146,7 @@
   final Completer<T> _inner;
 
   /// The callback to call if the future is canceled.
-  final FutureOrCallback _onCancel;
+  final FutureOrCallback? _onCancel;
 
   /// Creates a new completer for a [CancelableOperation].
   ///
@@ -155,15 +156,14 @@
   ///
   /// [onCancel] will be called synchronously when the operation is canceled.
   /// It's guaranteed to only be called once.
-  CancelableCompleter({FutureOr Function() onCancel})
+  CancelableCompleter({FutureOr Function()? onCancel})
       : _onCancel = onCancel,
         _inner = Completer<T>() {
-    _operation = CancelableOperation<T>._(this);
+    operation = CancelableOperation<T>._(this);
   }
 
   /// The operation controlled by this completer.
-  CancelableOperation<T> get operation => _operation;
-  CancelableOperation<T> _operation;
+  late final CancelableOperation<T> operation;
 
   /// Whether the completer has completed.
   bool get isCompleted => _isCompleted;
@@ -180,7 +180,7 @@
   ///
   /// If [value] is a [Future], this will complete to the result of that
   /// [Future] once it completes.
-  void complete([FutureOr<T> value]) {
+  void complete([FutureOr<T>? value]) {
     if (_isCompleted) throw StateError('Operation already completed');
     _isCompleted = true;
 
@@ -200,14 +200,14 @@
     future.then((result) {
       if (_isCanceled) return;
       _inner.complete(result);
-    }, onError: (error, StackTrace stackTrace) {
+    }, onError: (Object error, StackTrace stackTrace) {
       if (_isCanceled) return;
       _inner.completeError(error, stackTrace);
     });
   }
 
   /// Completes [operation] to [error].
-  void completeError(Object error, [StackTrace stackTrace]) {
+  void completeError(Object error, [StackTrace? stackTrace]) {
     if (_isCompleted) throw StateError('Operation already completed');
     _isCompleted = true;
 
@@ -221,7 +221,8 @@
 
     return _cancelMemo.runOnce(() {
       _isCanceled = true;
-      if (_onCancel != null) return _onCancel();
+      var onCancel = _onCancel;
+      if (onCancel != null) return onCancel();
     });
   }
 }
diff --git a/lib/src/delegate/event_sink.dart b/lib/src/delegate/event_sink.dart
index bc33b19..33d88e9 100644
--- a/lib/src/delegate/event_sink.dart
+++ b/lib/src/delegate/event_sink.dart
@@ -33,7 +33,7 @@
   }
 
   @override
-  void addError(error, [StackTrace stackTrace]) {
+  void addError(error, [StackTrace? stackTrace]) {
     _sink.addError(error, stackTrace);
   }
 
diff --git a/lib/src/delegate/future.dart b/lib/src/delegate/future.dart
index 984caf6..1155452 100644
--- a/lib/src/delegate/future.dart
+++ b/lib/src/delegate/future.dart
@@ -25,11 +25,11 @@
   Stream<T> asStream() => _future.asStream();
 
   @override
-  Future<T> catchError(Function onError, {bool Function(Object error) test}) =>
+  Future<T> catchError(Function onError, {bool Function(Object error)? test}) =>
       _future.catchError(onError, test: test);
 
   @override
-  Future<S> then<S>(FutureOr<S> Function(T) onValue, {Function onError}) =>
+  Future<S> then<S>(FutureOr<S> Function(T) onValue, {Function? onError}) =>
       _future.then(onValue, onError: onError);
 
   @override
@@ -37,6 +37,6 @@
       _future.whenComplete(action);
 
   @override
-  Future<T> timeout(Duration timeLimit, {FutureOr<T> Function() onTimeout}) =>
+  Future<T> timeout(Duration timeLimit, {FutureOr<T> Function()? onTimeout}) =>
       _future.timeout(timeLimit, onTimeout: onTimeout);
 }
diff --git a/lib/src/delegate/stream_sink.dart b/lib/src/delegate/stream_sink.dart
index 9005d1d..ad76e90 100644
--- a/lib/src/delegate/stream_sink.dart
+++ b/lib/src/delegate/stream_sink.dart
@@ -36,7 +36,7 @@
   }
 
   @override
-  void addError(error, [StackTrace stackTrace]) {
+  void addError(error, [StackTrace? stackTrace]) {
     _sink.addError(error, stackTrace);
   }
 
diff --git a/lib/src/delegate/stream_subscription.dart b/lib/src/delegate/stream_subscription.dart
index 392e27b..45b1134 100644
--- a/lib/src/delegate/stream_subscription.dart
+++ b/lib/src/delegate/stream_subscription.dart
@@ -31,22 +31,22 @@
           : TypeSafeStreamSubscription<T>(subscription);
 
   @override
-  void onData(void Function(T) handleData) {
+  void onData(void Function(T)? handleData) {
     _source.onData(handleData);
   }
 
   @override
-  void onError(Function handleError) {
+  void onError(Function? handleError) {
     _source.onError(handleError);
   }
 
   @override
-  void onDone(void Function() handleDone) {
+  void onDone(void Function()? handleDone) {
     _source.onDone(handleDone);
   }
 
   @override
-  void pause([Future resumeFuture]) {
+  void pause([Future? resumeFuture]) {
     _source.pause(resumeFuture);
   }
 
@@ -59,7 +59,7 @@
   Future cancel() => _source.cancel();
 
   @override
-  Future<E> asFuture<E>([E futureValue]) => _source.asFuture(futureValue);
+  Future<E> asFuture<E>([E? futureValue]) => _source.asFuture(futureValue);
 
   @override
   bool get isPaused => _source.isPaused;
diff --git a/lib/src/future_group.dart b/lib/src/future_group.dart
index 402ae46..3a6291f 100644
--- a/lib/src/future_group.dart
+++ b/lib/src/future_group.dart
@@ -44,13 +44,13 @@
   Stream get onIdle =>
       (_onIdleController ??= StreamController.broadcast(sync: true)).stream;
 
-  StreamController _onIdleController;
+  StreamController? _onIdleController;
 
   /// The values emitted by the futures that have been added to the group, in
   /// the order they were added.
   ///
   /// The slots for futures that haven't completed yet are `null`.
-  final _values = <T>[];
+  final _values = <T?>[];
 
   /// Wait for [task] to complete.
   @override
@@ -71,12 +71,13 @@
       _values[index] = value;
 
       if (_pending != 0) return null;
-      if (_onIdleController != null) _onIdleController.add(null);
+      var onIdleController = _onIdleController;
+      if (onIdleController != null) onIdleController.add(null);
 
       if (!_closed) return null;
-      if (_onIdleController != null) _onIdleController.close();
-      _completer.complete(_values);
-    }).catchError((error, StackTrace stackTrace) {
+      if (onIdleController != null) onIdleController.close();
+      _completer.complete(_values.whereType<T>().toList());
+    }).catchError((Object error, StackTrace stackTrace) {
       if (_completer.isCompleted) return null;
       _completer.completeError(error, stackTrace);
     });
@@ -89,6 +90,6 @@
     _closed = true;
     if (_pending != 0) return;
     if (_completer.isCompleted) return;
-    _completer.complete(_values);
+    _completer.complete(_values.whereType<T>().toList());
   }
 }
diff --git a/lib/src/lazy_stream.dart b/lib/src/lazy_stream.dart
index f5e65d1..e6779ee 100644
--- a/lib/src/lazy_stream.dart
+++ b/lib/src/lazy_stream.dart
@@ -15,7 +15,7 @@
 /// produce a `Stream`.
 class LazyStream<T> extends Stream<T> {
   /// The callback that's called to create the inner stream.
-  FutureOrCallback<Stream<T>> _callback;
+  FutureOrCallback<Stream<T>>? _callback;
 
   /// Creates a single-subscription `Stream` that calls [callback] when it gets
   /// a listener and forwards to the returned stream.
@@ -25,15 +25,15 @@
   }
 
   @override
-  StreamSubscription<T> listen(void Function(T) onData,
-      {Function onError, void Function() onDone, bool cancelOnError}) {
-    if (_callback == null) {
+  StreamSubscription<T> listen(void Function(T)? onData,
+      {Function? onError, void Function()? onDone, bool? cancelOnError}) {
+    var callback = _callback;
+    if (callback == null) {
       throw StateError('Stream has already been listened to.');
     }
 
     // Null out the callback before we invoke it to ensure that even while
     // running it, this can't be called twice.
-    var callback = _callback;
     _callback = null;
     var result = callback();
 
@@ -41,7 +41,7 @@
     if (result is Future<Stream<T>>) {
       stream = StreamCompleter.fromFuture(result);
     } else {
-      stream = result as Stream<T>;
+      stream = result;
     }
 
     return stream.listen(onData,
diff --git a/lib/src/null_stream_sink.dart b/lib/src/null_stream_sink.dart
index b28df2a..34b100b 100644
--- a/lib/src/null_stream_sink.dart
+++ b/lib/src/null_stream_sink.dart
@@ -43,12 +43,12 @@
   ///
   /// If [done] is passed, it's used as the [Sink.done] future. Otherwise, a
   /// completed future is used.
-  NullStreamSink({Future done}) : done = done ?? Future.value();
+  NullStreamSink({Future? done}) : done = done ?? 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])
+  NullStreamSink.error(Object error, [StackTrace? stackTrace])
       : done = 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
@@ -61,7 +61,7 @@
   }
 
   @override
-  void addError(error, [StackTrace stackTrace]) {
+  void addError(Object error, [StackTrace? stackTrace]) {
     _checkEventAllowed();
   }
 
@@ -70,7 +70,7 @@
     _checkEventAllowed();
 
     _addingStream = true;
-    var future = stream.listen(null).cancel() ?? Future.value();
+    var future = stream.listen(null).cancel();
     return future.whenComplete(() {
       _addingStream = false;
     });
diff --git a/lib/src/result/capture_sink.dart b/lib/src/result/capture_sink.dart
index a7c3a59..562f5f9 100644
--- a/lib/src/result/capture_sink.dart
+++ b/lib/src/result/capture_sink.dart
@@ -18,7 +18,7 @@
   }
 
   @override
-  void addError(Object error, [StackTrace stackTrace]) {
+  void addError(Object error, [StackTrace? stackTrace]) {
     _sink.add(Result.error(error, stackTrace));
   }
 
diff --git a/lib/src/result/error.dart b/lib/src/result/error.dart
index 76e0275..27ce155 100644
--- a/lib/src/result/error.dart
+++ b/lib/src/result/error.dart
@@ -8,7 +8,7 @@
 import 'value.dart';
 
 /// A result representing a thrown error.
-class ErrorResult implements Result<Null> {
+class ErrorResult implements Result<Never> {
   /// The error object that was thrown.
   final Object error;
 
@@ -20,11 +20,13 @@
   @override
   bool get isError => true;
   @override
-  ValueResult<Null> get asValue => null;
+  ValueResult<Never>? get asValue => null;
   @override
   ErrorResult get asError => this;
 
-  ErrorResult(this.error, this.stackTrace);
+  ErrorResult(this.error, [StackTrace? stackTrace])
+      // TODO: Use AsyncError.defaultStackTrace(error) once available
+      : stackTrace = stackTrace ?? StackTrace.fromString('');
 
   @override
   void complete(Completer completer) {
@@ -37,7 +39,7 @@
   }
 
   @override
-  Future<Null> get asFuture => Future<Null>.error(error, stackTrace);
+  Future<Never> get asFuture => Future<Never>.error(error, stackTrace);
 
   /// Calls an error handler with the error and stacktrace.
   ///
diff --git a/lib/src/result/future.dart b/lib/src/result/future.dart
index ff30546..20a5ebf 100644
--- a/lib/src/result/future.dart
+++ b/lib/src/result/future.dart
@@ -16,8 +16,8 @@
   /// The result of the wrapped [Future], if it's completed.
   ///
   /// If it hasn't completed yet, this will be `null`.
-  Result<T> get result => _result;
-  Result<T> _result;
+  Result<T>? get result => _result;
+  Result<T>? _result;
 
   ResultFuture(Future<T> future) : super(future) {
     Result.capture(future).then((result) {
diff --git a/lib/src/result/release_sink.dart b/lib/src/result/release_sink.dart
index 5d8267a..bf6dd50 100644
--- a/lib/src/result/release_sink.dart
+++ b/lib/src/result/release_sink.dart
@@ -18,7 +18,7 @@
   }
 
   @override
-  void addError(Object error, [StackTrace stackTrace]) {
+  void addError(Object error, [StackTrace? stackTrace]) {
     // Errors may be added by intermediate processing, even if it is never
     // added by CaptureSink.
     _sink.addError(error, stackTrace);
diff --git a/lib/src/result/result.dart b/lib/src/result/result.dart
index e604782..5f93b55 100644
--- a/lib/src/result/result.dart
+++ b/lib/src/result/result.dart
@@ -63,7 +63,7 @@
   factory Result(T Function() computation) {
     try {
       return ValueResult<T>(computation());
-    } catch (e, s) {
+    } on Object catch (e, s) {
       return ErrorResult(e, s);
     }
   }
@@ -76,7 +76,7 @@
   /// Creates a `Result` holding an error.
   ///
   /// Alias for [ErrorResult.ErrorResult].
-  factory Result.error(Object error, [StackTrace stackTrace]) =>
+  factory Result.error(Object error, [StackTrace? stackTrace]) =>
       ErrorResult(error, stackTrace);
 
   /// Captures the result of a future into a `Result` future.
@@ -85,7 +85,7 @@
   /// Errors have been converted to an [ErrorResult] value.
   static Future<Result<T>> capture<T>(Future<T> future) {
     return future.then((value) => ValueResult(value),
-        onError: (error, StackTrace stackTrace) =>
+        onError: (Object error, StackTrace stackTrace) =>
             ErrorResult(error, stackTrace));
   }
 
@@ -97,9 +97,9 @@
   /// wrapped as a [Result.value].
   /// The returned future will never have an error.
   static Future<List<Result<T>>> captureAll<T>(Iterable<FutureOr<T>> elements) {
-    var results = <Result<T>>[];
+    var results = <Result<T>?>[];
     var pending = 0;
-    Completer<List<Result<T>>> completer;
+    late Completer<List<Result<T>>> completer;
     for (var element in elements) {
       if (element is Future<T>) {
         var i = results.length;
@@ -108,15 +108,15 @@
         Result.capture<T>(element).then((result) {
           results[i] = result;
           if (--pending == 0) {
-            completer.complete(results);
+            completer.complete(List.from(results));
           }
         });
       } else {
-        results.add(Result<T>.value(element as T));
+        results.add(Result<T>.value(element));
       }
     }
     if (pending == 0) {
-      return Future<List<Result<T>>>.value(results);
+      return Future.value(List.from(results));
     }
     completer = Completer<List<Result<T>>>();
     return completer.future;
@@ -172,8 +172,8 @@
   /// Otherwise both levels of results are value results, and a single
   /// result with the value is returned.
   static Result<T> flatten<T>(Result<Result<T>> result) {
-    if (result.isValue) return result.asValue.value;
-    return result.asError;
+    if (result.isValue) return result.asValue!.value;
+    return result.asError!;
   }
 
   /// Converts a sequence of results to a result of a list.
@@ -184,9 +184,9 @@
     var values = <T>[];
     for (var result in results) {
       if (result.isValue) {
-        values.add(result.asValue.value);
+        values.add(result.asValue!.value);
       } else {
-        return result.asError;
+        return result.asError!;
       }
     }
     return Result<List<T>>.value(values);
@@ -205,12 +205,12 @@
   /// If this is a value result, returns itself.
   ///
   /// Otherwise returns `null`.
-  ValueResult<T> get asValue;
+  ValueResult<T>? get asValue;
 
   /// If this is an error result, returns itself.
   ///
   /// Otherwise returns `null`.
-  ErrorResult get asError;
+  ErrorResult? get asError;
 
   /// Completes a completer with this result.
   void complete(Completer<T> completer);
diff --git a/lib/src/result/value.dart b/lib/src/result/value.dart
index 5c1a60f..7cfc474 100644
--- a/lib/src/result/value.dart
+++ b/lib/src/result/value.dart
@@ -19,7 +19,7 @@
   @override
   ValueResult<T> get asValue => this;
   @override
-  ErrorResult get asError => null;
+  ErrorResult? get asError => null;
 
   ValueResult(this.value);
 
diff --git a/lib/src/single_subscription_transformer.dart b/lib/src/single_subscription_transformer.dart
index fe939fc..2e056b2 100644
--- a/lib/src/single_subscription_transformer.dart
+++ b/lib/src/single_subscription_transformer.dart
@@ -11,14 +11,14 @@
 /// listening to a stream as soon as it's bound.
 ///
 /// This also casts the source stream's events to type `T`. If the cast fails,
-/// the result stream will emit a [CastError]. This behavior is deprecated, and
+/// the result stream will emit a [TypeError]. This behavior is deprecated, and
 /// should not be relied upon.
 class SingleSubscriptionTransformer<S, T> extends StreamTransformerBase<S, T> {
   const SingleSubscriptionTransformer();
 
   @override
   Stream<T> bind(Stream<S> stream) {
-    StreamSubscription<S> subscription;
+    late StreamSubscription<S> subscription;
     var controller =
         StreamController<T>(sync: true, onCancel: () => subscription.cancel());
     subscription = stream.listen((value) {
@@ -26,7 +26,7 @@
       // type parameter and avoid this conversion.
       try {
         controller.add(value as T);
-      } on CastError catch (error, stackTrace) {
+      } on TypeError catch (error, stackTrace) {
         controller.addError(error, stackTrace);
       }
     }, onError: controller.addError, onDone: controller.close);
diff --git a/lib/src/stream_completer.dart b/lib/src/stream_completer.dart
index 5128d57..27034c2 100644
--- a/lib/src/stream_completer.dart
+++ b/lib/src/stream_completer.dart
@@ -97,7 +97,7 @@
   ///
   /// 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]) {
+  void setError(Object error, [StackTrace? stackTrace]) {
     setSourceStream(Stream.fromFuture(Future.error(error, stackTrace)));
   }
 }
@@ -108,30 +108,31 @@
   ///
   /// Created if the user listens on this stream before the source stream
   /// is set, or if using [_setEmpty] so there is no source stream.
-  StreamController<T> _controller;
+  StreamController<T>? _controller;
 
   /// Source stream for the events provided by this stream.
   ///
   /// Set when the completer sets the source stream using [_setSourceStream]
   /// or [_setEmpty].
-  Stream<T> _sourceStream;
+  Stream<T>? _sourceStream;
 
   @override
-  StreamSubscription<T> listen(void Function(T) onData,
-      {Function onError, void Function() onDone, bool cancelOnError}) {
+  StreamSubscription<T> listen(void Function(T)? onData,
+      {Function? onError, void Function()? onDone, bool? cancelOnError}) {
     if (_controller == null) {
-      if (_sourceStream != null && !_sourceStream.isBroadcast) {
+      var sourceStream = _sourceStream;
+      if (sourceStream != null && !sourceStream.isBroadcast) {
         // If the source stream is itself single subscription,
         // just listen to it directly instead of creating a controller.
-        return _sourceStream.listen(onData,
+        return sourceStream.listen(onData,
             onError: onError, onDone: onDone, cancelOnError: cancelOnError);
       }
-      _createController();
+      _ensureController();
       if (_sourceStream != null) {
         _linkStreamToController();
       }
     }
-    return _controller.stream.listen(onData,
+    return _controller!.stream.listen(onData,
         onError: onError, onDone: onDone, cancelOnError: cancelOnError);
   }
 
@@ -157,11 +158,10 @@
 
   /// Links source stream to controller when both are available.
   void _linkStreamToController() {
-    assert(_controller != null);
-    assert(_sourceStream != null);
-    _controller
-        .addStream(_sourceStream, cancelOnError: false)
-        .whenComplete(_controller.close);
+    var controller = _controller!;
+    controller
+        .addStream(_sourceStream!, cancelOnError: false)
+        .whenComplete(controller.close);
   }
 
   /// Sets an empty source stream.
@@ -170,16 +170,13 @@
   /// immediately.
   void _setEmpty() {
     assert(_sourceStream == null);
-    if (_controller == null) {
-      _createController();
-    }
-    _sourceStream = _controller.stream; // Mark stream as set.
-    _controller.close();
+    var controller = _ensureController();
+    _sourceStream = controller.stream; // Mark stream as set.
+    controller.close();
   }
 
   // Creates the [_controller].
-  void _createController() {
-    assert(_controller == null);
-    _controller = StreamController<T>(sync: true);
+  StreamController<T> _ensureController() {
+    return _controller ??= StreamController<T>(sync: true);
   }
 }
diff --git a/lib/src/stream_group.dart b/lib/src/stream_group.dart
index e9a4afc..8c17ec2 100644
--- a/lib/src/stream_group.dart
+++ b/lib/src/stream_group.dart
@@ -29,7 +29,7 @@
 class StreamGroup<T> implements Sink<Stream<T>> {
   /// The stream through which all events from streams in the group are emitted.
   Stream<T> get stream => _controller.stream;
-  StreamController<T> _controller;
+  late StreamController<T> _controller;
 
   /// Whether the group is closed, meaning that no more streams may be added.
   var _closed = false;
@@ -47,7 +47,7 @@
   /// subscriptions will be canceled and set to null again. Single-subscriber
   /// stream subscriptions will be left intact, since they can't be
   /// re-subscribed.
-  final _subscriptions = <Stream<T>, StreamSubscription<T>>{};
+  final _subscriptions = <Stream<T>, StreamSubscription<T>?>{};
 
   /// Merges the events from [streams] into a single single-subscription stream.
   ///
@@ -100,7 +100,7 @@
   ///
   /// Throws a [StateError] if this group is closed.
   @override
-  Future add(Stream<T> stream) {
+  Future? add(Stream<T> stream) {
     if (_closed) {
       throw StateError("Can't add a Stream to a closed StreamGroup.");
     }
@@ -130,7 +130,7 @@
   ///
   /// If [stream]'s subscription is canceled, this returns
   /// [StreamSubscription.cancel]'s return value. Otherwise, it returns `null`.
-  Future remove(Stream<T> stream) {
+  Future? remove(Stream<T> stream) {
     var subscription = _subscriptions.remove(stream);
     var future = subscription == null ? null : subscription.cancel();
     if (_closed && _subscriptions.isEmpty) _controller.close();
@@ -155,7 +155,7 @@
   void _onPause() {
     _state = _StreamGroupState.paused;
     for (var subscription in _subscriptions.values) {
-      subscription.pause();
+      subscription!.pause();
     }
   }
 
@@ -163,19 +163,18 @@
   void _onResume() {
     _state = _StreamGroupState.listening;
     for (var subscription in _subscriptions.values) {
-      subscription.resume();
+      subscription!.resume();
     }
   }
 
   /// A callback called when [stream] is canceled.
   ///
   /// This is only called for single-subscription groups.
-  Future _onCancel() {
+  Future? _onCancel() {
     _state = _StreamGroupState.canceled;
 
     var futures = _subscriptions.values
-        .map((subscription) => subscription.cancel())
-        .where((future) => future != null)
+        .map((subscription) => subscription!.cancel())
         .toList();
 
     _subscriptions.clear();
@@ -194,7 +193,7 @@
       // will still be added to [_controller], but then they'll be dropped since
       // it has no listeners.
       if (!stream.isBroadcast) return;
-      subscription.cancel();
+      subscription!.cancel();
       _subscriptions[stream] = null;
     });
   }
diff --git a/lib/src/stream_queue.dart b/lib/src/stream_queue.dart
index 278e249..5aa6054 100644
--- a/lib/src/stream_queue.dart
+++ b/lib/src/stream_queue.dart
@@ -87,7 +87,7 @@
   ///
   /// Set to subscription when listening, and set to `null` when the
   /// subscription is done (and [_isDone] is set to true).
-  StreamSubscription<T> _subscription;
+  StreamSubscription<T>? _subscription;
 
   /// Whether the event source is done.
   bool _isDone = false;
@@ -343,7 +343,7 @@
         transaction.reject();
       }
       return result;
-    }, onError: (error) {
+    }, onError: (Object error) {
       transaction.commit(queue);
       throw error;
     });
@@ -399,7 +399,7 @@
   /// After calling `cancel`, no further events can be requested.
   /// None of [lookAhead], [next], [peek], [rest], [skip], [take] or [cancel]
   /// may be called again.
-  Future cancel({bool immediate = false}) {
+  Future? cancel({bool immediate = false}) {
     if (_isClosed) throw _failClosed();
     _isClosed = true;
 
@@ -454,11 +454,10 @@
     }
     _isDone = true;
 
-    if (_subscription == null) {
+    var subscription = _subscription;
+    if (subscription == null) {
       return _source;
     }
-
-    var subscription = _subscription;
     _subscription = null;
 
     var wasPaused = subscription.isPaused;
@@ -475,7 +474,7 @@
   ///
   /// The event source is restarted by the next call to [_ensureListening].
   void _pause() {
-    _subscription.pause();
+    _subscription!.pause();
   }
 
   /// Ensures that we are listening on events from the event source.
@@ -488,22 +487,22 @@
     if (_subscription == null) {
       _subscription = _source.listen((data) {
         _addResult(Result.value(data));
-      }, onError: (error, StackTrace stackTrace) {
+      }, onError: (Object error, StackTrace stackTrace) {
         _addResult(Result.error(error, stackTrace));
       }, onDone: () {
         _subscription = null;
         _close();
       });
     } else {
-      _subscription.resume();
+      _subscription!.resume();
     }
   }
 
   /// Cancels the underlying event source.
-  Future _cancel() {
+  Future? _cancel() {
     if (_isDone) return null;
     _subscription ??= _source.listen(null);
-    var future = _subscription.cancel();
+    var future = _subscription!.cancel();
     _close();
     return future;
   }
@@ -771,7 +770,8 @@
 
       var event = events.removeFirst();
       if (event.isError) {
-        _completer.completeError(event.asError.error, event.asError.stackTrace);
+        _completer.completeError(
+            event.asError!.error, event.asError!.stackTrace);
         return true;
       }
     }
@@ -814,10 +814,10 @@
 
       var event = events.removeFirst();
       if (event.isError) {
-        event.asError.complete(_completer);
+        event.asError!.complete(_completer);
         return true;
       }
-      _list.add(event.asValue.value);
+      _list.add(event.asValue!.value);
     }
     _completer.complete(_list);
     return true;
@@ -837,10 +837,10 @@
       }
       var event = events.elementAt(_list.length);
       if (event.isError) {
-        event.asError.complete(_completer);
+        event.asError!.complete(_completer);
         return true;
       }
-      _list.add(event.asValue.value);
+      _list.add(event.asValue!.value);
     }
     _completer.complete(_list);
     return true;
@@ -954,8 +954,7 @@
 /// [StreamQueue._updateRequests].
 class _TransactionRequest<T> implements _EventRequest<T> {
   /// The transaction created by this request.
-  StreamQueueTransaction<T> get transaction => _transaction;
-  StreamQueueTransaction<T> _transaction;
+  late final StreamQueueTransaction<T> transaction;
 
   /// The controller that passes events to [transaction].
   final _controller = StreamController<T>(sync: true);
@@ -964,7 +963,7 @@
   var _eventsSent = 0;
 
   _TransactionRequest(StreamQueue<T> parent) {
-    _transaction = StreamQueueTransaction._(parent, _controller.stream);
+    transaction = StreamQueueTransaction._(parent, _controller.stream);
   }
 
   @override
@@ -973,6 +972,6 @@
       events[_eventsSent++].addTo(_controller);
     }
     if (isDone && !_controller.isClosed) _controller.close();
-    return transaction._committed || _transaction._rejected;
+    return transaction._committed || transaction._rejected;
   }
 }
diff --git a/lib/src/stream_sink_completer.dart b/lib/src/stream_sink_completer.dart
index ebfd717..10e549d 100644
--- a/lib/src/stream_sink_completer.dart
+++ b/lib/src/stream_sink_completer.dart
@@ -70,7 +70,7 @@
   ///
   /// 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]) {
+  void setError(Object error, [StackTrace? stackTrace]) {
     setDestinationSink(NullStreamSink.error(error, stackTrace));
   }
 }
@@ -81,18 +81,18 @@
   ///
   /// Created if the user adds events to this sink before the destination sink
   /// is set.
-  StreamController<T> _controller;
+  StreamController<T>? _controller;
 
   /// Completer for [done].
   ///
   /// Created if the user requests the [done] future before the destination sink
   /// is set.
-  Completer _doneCompleter;
+  Completer? _doneCompleter;
 
   /// Destination sink for the events added to this sink.
   ///
   /// Set when [StreamSinkCompleter.setDestinationSink] is called.
-  StreamSink<T> _destinationSink;
+  StreamSink<T>? _destinationSink;
 
   /// Whether events should be sent directly to [_destinationSink], as opposed
   /// to going through [_controller].
@@ -100,56 +100,52 @@
 
   @override
   Future get done {
-    if (_doneCompleter != null) return _doneCompleter.future;
+    if (_doneCompleter != null) return _doneCompleter!.future;
     if (_destinationSink == null) {
       _doneCompleter = Completer.sync();
-      return _doneCompleter.future;
+      return _doneCompleter!.future;
     }
-    return _destinationSink.done;
+    return _destinationSink!.done;
   }
 
   @override
   void add(T event) {
     if (_canSendDirectly) {
-      _destinationSink.add(event);
+      _destinationSink!.add(event);
     } else {
-      _ensureController();
-      _controller.add(event);
+      _ensureController().add(event);
     }
   }
 
   @override
-  void addError(error, [StackTrace stackTrace]) {
+  void addError(error, [StackTrace? stackTrace]) {
     if (_canSendDirectly) {
-      _destinationSink.addError(error, stackTrace);
+      _destinationSink!.addError(error, stackTrace);
     } else {
-      _ensureController();
-      _controller.addError(error, stackTrace);
+      _ensureController().addError(error, stackTrace);
     }
   }
 
   @override
   Future addStream(Stream<T> stream) {
-    if (_canSendDirectly) return _destinationSink.addStream(stream);
+    if (_canSendDirectly) return _destinationSink!.addStream(stream);
 
-    _ensureController();
-    return _controller.addStream(stream, cancelOnError: false);
+    return _ensureController().addStream(stream, cancelOnError: false);
   }
 
   @override
   Future close() {
     if (_canSendDirectly) {
-      _destinationSink.close();
+      _destinationSink!.close();
     } else {
-      _ensureController();
-      _controller.close();
+      _ensureController().close();
     }
     return done;
   }
 
   /// Create [_controller] if it doesn't yet exist.
-  void _ensureController() {
-    _controller ??= StreamController(sync: true);
+  StreamController<T> _ensureController() {
+    return _controller ??= StreamController(sync: true);
   }
 
   /// Sets the destination sink to which events from this sink will be provided.
@@ -168,7 +164,7 @@
       // Catch any error that may come from [addStream] or [sink.close]. They'll
       // be reported through [done] anyway.
       sink
-          .addStream(_controller.stream)
+          .addStream(_controller!.stream)
           .whenComplete(sink.close)
           .catchError((_) {});
     }
@@ -176,7 +172,7 @@
     // If the user has already asked when the sink is done, connect the sink's
     // done callback to that completer.
     if (_doneCompleter != null) {
-      _doneCompleter.complete(sink.done);
+      _doneCompleter!.complete(sink.done);
     }
   }
 }
diff --git a/lib/src/stream_sink_transformer.dart b/lib/src/stream_sink_transformer.dart
index 1dc27ed..bdcb196 100644
--- a/lib/src/stream_sink_transformer.dart
+++ b/lib/src/stream_sink_transformer.dart
@@ -34,9 +34,9 @@
   /// they're passed are forwarded to the inner sink. If a handler is omitted,
   /// the event is passed through unaltered.
   factory StreamSinkTransformer.fromHandlers(
-      {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}) {
     return HandlerTransformer<S, T>(handleData, handleError, handleDone);
   }
 
diff --git a/lib/src/stream_sink_transformer/handler_transformer.dart b/lib/src/stream_sink_transformer/handler_transformer.dart
index 8d37ce3..b112e3e 100644
--- a/lib/src/stream_sink_transformer/handler_transformer.dart
+++ b/lib/src/stream_sink_transformer/handler_transformer.dart
@@ -11,6 +11,9 @@
 typedef HandleData<S, T> = void Function(S data, EventSink<T> sink);
 
 /// The type of the callback for handling error events.
+//
+// TODO: Update to take a non-nullable StackTrace once that change lands in
+// the sdk.
 typedef HandleError<T> = void Function(
     Object error, StackTrace stackTrace, EventSink<T> sink);
 
@@ -20,13 +23,13 @@
 /// A [StreamSinkTransformer] that delegates events to the given handlers.
 class HandlerTransformer<S, T> implements StreamSinkTransformer<S, T> {
   /// The handler for data events.
-  final HandleData<S, T> _handleData;
+  final HandleData<S, T>? _handleData;
 
   /// The handler for error events.
-  final HandleError<T> _handleError;
+  final HandleError<T>? _handleError;
 
   /// The handler for done events.
-  final HandleDone<T> _handleDone;
+  final HandleDone<T>? _handleDone;
 
   HandlerTransformer(this._handleData, this._handleError, this._handleDone);
 
@@ -55,19 +58,22 @@
 
   @override
   void add(S event) {
-    if (_transformer._handleData == null) {
+    var handleData = _transformer._handleData;
+    if (handleData == null) {
       _inner.add(event as T);
     } else {
-      _transformer._handleData(event, _safeCloseInner);
+      handleData(event, _safeCloseInner);
     }
   }
 
   @override
-  void addError(error, [StackTrace stackTrace]) {
-    if (_transformer._handleError == null) {
+  void addError(error, [StackTrace? stackTrace]) {
+    var handleError = _transformer._handleError;
+    if (handleError == null) {
       _inner.addError(error, stackTrace);
     } else {
-      _transformer._handleError(error, stackTrace, _safeCloseInner);
+      handleError(error, stackTrace ?? AsyncError.defaultStackTrace(error),
+          _safeCloseInner);
     }
   }
 
@@ -82,9 +88,10 @@
 
   @override
   Future close() {
-    if (_transformer._handleDone == null) return _inner.close();
+    var handleDone = _transformer._handleDone;
+    if (handleDone == null) return _inner.close();
 
-    _transformer._handleDone(_safeCloseInner);
+    handleDone(_safeCloseInner);
     return _inner.done;
   }
 }
diff --git a/lib/src/stream_sink_transformer/stream_transformer_wrapper.dart b/lib/src/stream_sink_transformer/stream_transformer_wrapper.dart
index 1df7e5a..2afcbff 100644
--- a/lib/src/stream_sink_transformer/stream_transformer_wrapper.dart
+++ b/lib/src/stream_sink_transformer/stream_transformer_wrapper.dart
@@ -50,7 +50,7 @@
   }
 
   @override
-  void addError(error, [StackTrace stackTrace]) {
+  void addError(error, [StackTrace? stackTrace]) {
     _controller.addError(error, stackTrace);
   }
 
diff --git a/lib/src/stream_splitter.dart b/lib/src/stream_splitter.dart
index e9f326e..f7377d6 100644
--- a/lib/src/stream_splitter.dart
+++ b/lib/src/stream_splitter.dart
@@ -28,7 +28,7 @@
   /// The subscription to [_stream].
   ///
   /// This will be `null` until a branch has a listener.
-  StreamSubscription<T> _subscription;
+  StreamSubscription<T>? _subscription;
 
   /// The buffer of events or errors that have already been emitted by
   /// [_stream].
@@ -57,7 +57,7 @@
   ///
   /// [count] defaults to 2. This is the same as creating [count] branches and
   /// then closing the [StreamSplitter].
-  static List<Stream<T>> splitFrom<T>(Stream<T> stream, [int count]) {
+  static List<Stream<T>> splitFrom<T>(Stream<T> stream, [int? count]) {
     count ??= 2;
     var splitter = StreamSplitter<T>(stream);
     var streams = List<Stream<T>>.generate(count, (_) => splitter.split());
@@ -125,8 +125,8 @@
     assert(_controllers.isEmpty);
     assert(_isClosed);
 
-    Future future;
-    if (_subscription != null) future = _subscription.cancel();
+    Future? future;
+    if (_subscription != null) future = _subscription!.cancel();
     if (future != null) _closeGroup.add(future);
     _closeGroup.close();
   }
@@ -142,7 +142,7 @@
       // Resume the subscription in case it was paused, either because all the
       // controllers were paused or because the last one was canceled. If it
       // wasn't paused, this will be a no-op.
-      _subscription.resume();
+      _subscription!.resume();
     } else {
       _subscription =
           _stream.listen(_onData, onError: _onError, onDone: _onDone);
@@ -152,14 +152,14 @@
   /// Pauses [_subscription] if every controller is paused.
   void _onPause() {
     if (!_controllers.every((controller) => controller.isPaused)) return;
-    _subscription.pause();
+    _subscription!.pause();
   }
 
   /// Resumes [_subscription].
   ///
   /// If [_subscription] wasn't paused, this is a no-op.
   void _onResume() {
-    _subscription.resume();
+    _subscription!.resume();
   }
 
   /// Removes [controller] from [_controllers] and cancels or pauses
@@ -175,7 +175,7 @@
     if (_isClosed) {
       _cancelSubscription();
     } else {
-      _subscription.pause();
+      _subscription!.pause();
     }
   }
 
diff --git a/lib/src/stream_subscription_transformer.dart b/lib/src/stream_subscription_transformer.dart
index 0e28972..d03ea70 100644
--- a/lib/src/stream_subscription_transformer.dart
+++ b/lib/src/stream_subscription_transformer.dart
@@ -28,9 +28,9 @@
 /// [StreamSubscription]: [handleCancel] must call `cancel()`, [handlePause]
 /// must call `pause()`, and [handleResume] must call `resume()`.
 StreamTransformer<T, T> subscriptionTransformer<T>(
-    {Future Function(StreamSubscription<T>) handleCancel,
-    void Function(StreamSubscription<T>) handlePause,
-    void Function(StreamSubscription<T>) handleResume}) {
+    {Future Function(StreamSubscription<T>)? handleCancel,
+    void Function(StreamSubscription<T>)? handlePause,
+    void Function(StreamSubscription<T>)? handleResume}) {
   return StreamTransformer((stream, cancelOnError) {
     return _TransformedSubscription(
         stream.listen(null, cancelOnError: cancelOnError),
@@ -50,7 +50,7 @@
 /// methods.
 class _TransformedSubscription<T> implements StreamSubscription<T> {
   /// The wrapped subscription.
-  StreamSubscription<T> _inner;
+  StreamSubscription<T>? _inner;
 
   /// The callback to run when [cancel] is called.
   final _AsyncHandler<T> _handleCancel;
@@ -68,47 +68,47 @@
       this._inner, this._handleCancel, this._handlePause, this._handleResume);
 
   @override
-  void onData(void Function(T) handleData) {
+  void onData(void Function(T)? handleData) {
     _inner?.onData(handleData);
   }
 
   @override
-  void onError(Function handleError) {
+  void onError(Function? handleError) {
     _inner?.onError(handleError);
   }
 
   @override
-  void onDone(void Function() handleDone) {
+  void onDone(void Function()? handleDone) {
     _inner?.onDone(handleDone);
   }
 
   @override
   Future cancel() => _cancelMemoizer.runOnce(() {
-        var inner = _inner;
-        _inner.onData(null);
-        _inner.onDone(null);
+        var inner = _inner!;
+        inner.onData(null);
+        inner.onDone(null);
 
         // Setting onError to null will cause errors to be top-leveled.
-        _inner.onError((_, __) {});
+        inner.onError((_, __) {});
         _inner = null;
         return _handleCancel(inner);
       });
   final _cancelMemoizer = AsyncMemoizer();
 
   @override
-  void pause([Future resumeFuture]) {
+  void pause([Future? resumeFuture]) {
     if (_cancelMemoizer.hasRun) return;
     if (resumeFuture != null) resumeFuture.whenComplete(resume);
-    _handlePause(_inner);
+    _handlePause(_inner!);
   }
 
   @override
   void resume() {
     if (_cancelMemoizer.hasRun) return;
-    _handleResume(_inner);
+    _handleResume(_inner!);
   }
 
   @override
-  Future<E> asFuture<E>([E futureValue]) =>
+  Future<E> asFuture<E>([E? futureValue]) =>
       _inner?.asFuture(futureValue) ?? Completer<E>().future;
 }
diff --git a/lib/src/stream_zip.dart b/lib/src/stream_zip.dart
index e319746..506bf6d 100644
--- a/lib/src/stream_zip.dart
+++ b/lib/src/stream_zip.dart
@@ -18,12 +18,12 @@
   StreamZip(Iterable<Stream<T>> streams) : _streams = streams;
 
   @override
-  StreamSubscription<List<T>> listen(void Function(List<T>) onData,
-      {Function onError, void Function() onDone, bool cancelOnError}) {
+  StreamSubscription<List<T>> listen(void Function(List<T>)? onData,
+      {Function? onError, void Function()? onDone, bool? cancelOnError}) {
     cancelOnError = identical(true, cancelOnError);
     var subscriptions = <StreamSubscription<T>>[];
-    StreamController<List<T>> controller;
-    List<T> current;
+    late StreamController<List<T>> controller;
+    late List<T?> current;
     var dataCount = 0;
 
     /// Called for each data from a subscription in [subscriptions].
@@ -31,8 +31,8 @@
       current[index] = data;
       dataCount++;
       if (dataCount == subscriptions.length) {
-        var data = current;
-        current = List(subscriptions.length);
+        var data = List<T>.from(current);
+        current = List<T?>.filled(subscriptions.length, null);
         dataCount = 0;
         for (var i = 0; i < subscriptions.length; i++) {
           if (i != index) subscriptions[i].resume();
@@ -85,7 +85,7 @@
       rethrow;
     }
 
-    current = List(subscriptions.length);
+    current = List<T?>.filled(subscriptions.length, null);
 
     controller = StreamController<List<T>>(onPause: () {
       for (var i = 0; i < subscriptions.length; i++) {
diff --git a/lib/src/subscription_stream.dart b/lib/src/subscription_stream.dart
index b428619..2c94e05 100644
--- a/lib/src/subscription_stream.dart
+++ b/lib/src/subscription_stream.dart
@@ -18,7 +18,7 @@
 /// If other code is accessing the subscription, results may be unpredictable.
 class SubscriptionStream<T> extends Stream<T> {
   /// The subscription providing the events for this stream.
-  StreamSubscription<T> _source;
+  StreamSubscription<T>? _source;
 
   /// Create a single-subscription `Stream` from [subscription].
   ///
@@ -31,21 +31,22 @@
   /// an error.
   SubscriptionStream(StreamSubscription<T> subscription)
       : _source = subscription {
-    _source.pause();
+    var source = _source!;
+    source.pause();
     // Clear callbacks to avoid keeping them alive unnecessarily.
-    _source.onData(null);
-    _source.onError(null);
-    _source.onDone(null);
+    source.onData(null);
+    source.onError(null);
+    source.onDone(null);
   }
 
   @override
-  StreamSubscription<T> listen(void Function(T) onData,
-      {Function onError, void Function() onDone, bool cancelOnError}) {
-    if (_source == null) {
+  StreamSubscription<T> listen(void Function(T)? onData,
+      {Function? onError, void Function()? onDone, bool? cancelOnError}) {
+    var subscription = _source;
+    if (subscription == null) {
       throw StateError('Stream has already been listened to.');
     }
     cancelOnError = (true == cancelOnError);
-    var subscription = _source;
     _source = null;
 
     var result = cancelOnError
@@ -71,26 +72,17 @@
       : super(subscription);
 
   @override
-  void onError(Function handleError) {
+  void onError(Function? handleError) {
     // Cancel when receiving an error.
     super.onError((error, StackTrace stackTrace) {
-      var cancelFuture = super.cancel();
-      if (cancelFuture != null) {
-        // Wait for the cancel to complete before sending the error event.
-        cancelFuture.whenComplete(() {
-          if (handleError is ZoneBinaryCallback) {
-            handleError(error, stackTrace);
-          } else {
-            handleError(error);
-          }
-        });
-      } else {
+      // Wait for the cancel to complete before sending the error event.
+      super.cancel().whenComplete(() {
         if (handleError is ZoneBinaryCallback) {
           handleError(error, stackTrace);
-        } else {
+        } else if (handleError != null) {
           handleError(error);
         }
-      }
+      });
     });
   }
 }
diff --git a/lib/src/typed/stream_subscription.dart b/lib/src/typed/stream_subscription.dart
index d85b776..fe91656 100644
--- a/lib/src/typed/stream_subscription.dart
+++ b/lib/src/typed/stream_subscription.dart
@@ -13,22 +13,23 @@
   TypeSafeStreamSubscription(this._subscription);
 
   @override
-  void onData(void Function(T) handleData) {
+  void onData(void Function(T)? handleData) {
+    if (handleData == null) return _subscription.onData(null);
     _subscription.onData((data) => handleData(data as T));
   }
 
   @override
-  void onError(Function handleError) {
+  void onError(Function? handleError) {
     _subscription.onError(handleError);
   }
 
   @override
-  void onDone(void Function() handleDone) {
+  void onDone(void Function()? handleDone) {
     _subscription.onDone(handleDone);
   }
 
   @override
-  void pause([Future resumeFuture]) {
+  void pause([Future? resumeFuture]) {
     _subscription.pause(resumeFuture);
   }
 
@@ -41,5 +42,6 @@
   Future cancel() => _subscription.cancel();
 
   @override
-  Future<E> asFuture<E>([E futureValue]) => _subscription.asFuture(futureValue);
+  Future<E> asFuture<E>([E? futureValue]) =>
+      _subscription.asFuture(futureValue);
 }
diff --git a/pubspec.yaml b/pubspec.yaml
index 0cb71cf..bfdf20e 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -1,11 +1,11 @@
 name: async
-version: 2.4.2
+version: 2.5.0-nullsafety
 
 description: Utility functions and classes related to the 'dart:async' library.
 homepage: https://www.github.com/dart-lang/async
 
 environment:
-  sdk: '>=2.2.0 <3.0.0'
+  sdk: '>=2.9.0-20.0.dev <2.9.0'
 
 dependencies:
   collection: ^1.5.0
@@ -15,3 +15,81 @@
   stack_trace: ^1.0.0
   test: ^1.0.0
   pedantic: ^1.0.0
+
+dependency_overrides:
+  boolean_selector:
+    git:
+      url: git://github.com/dart-lang/boolean_selector.git
+      ref: null_safety
+  charcode:
+    git:
+      url: git://github.com/dart-lang/charcode.git
+      ref: null_safety
+  clock:
+    git:
+      url: git://github.com/dart-lang/clock.git
+      ref: null_safety
+  collection:
+    git:
+      url: git://github.com/dart-lang/collection.git
+      ref: null_safety
+  fake_async:
+    git:
+      url: git://github.com/dart-lang/fake_async.git
+      ref: null_safety
+  matcher:
+    git:
+      url: git://github.com/dart-lang/matcher.git
+      ref: null_safety
+  meta:
+    git:
+      url: git://github.com/dart-lang/sdk.git
+      ref: null_safety-pkgs
+      path: pkg/meta
+  path:
+    git:
+      url: git://github.com/dart-lang/path.git
+      ref: null_safety
+  pedantic:
+    git:
+      url: git://github.com/dart-lang/pedantic.git
+      ref: null_safety
+  pool:
+    git:
+      url: git://github.com/dart-lang/pool.git
+      ref: null_safety
+  source_span:
+    git:
+      url: git://github.com/dart-lang/source_span.git
+      ref: null_safety
+  stack_trace:
+    git:
+      url: git://github.com/dart-lang/stack_trace.git
+      ref: null_safety
+  stream_channel:
+    git:
+      url: git://github.com/dart-lang/stream_channel.git
+      ref: null_safety
+  string_scanner:
+    git:
+      url: git://github.com/dart-lang/string_scanner.git
+      ref: null_safety
+  term_glyph:
+    git:
+      url: git://github.com/dart-lang/term_glyph.git
+      ref: null_safety
+  test_api:
+    git:
+      url: git://github.com/dart-lang/test.git
+      ref: null_safety
+      path: pkgs/test_api
+  test_core:
+    git:
+      url: git://github.com/dart-lang/test.git
+      ref: null_safety
+      path: pkgs/test_core
+  test:
+    git:
+      url: git://github.com/dart-lang/test.git
+      ref: null_safety
+      path: pkgs/test
diff --git a/test/async_cache_test.dart b/test/async_cache_test.dart
index 4efbb69..81e43e1 100644
--- a/test/async_cache_test.dart
+++ b/test/async_cache_test.dart
@@ -9,7 +9,7 @@
 import 'package:test/test.dart';
 
 void main() {
-  AsyncCache<String> cache;
+  late AsyncCache<String> cache;
 
   setUp(() {
     // Create a cache that is fresh for an hour.
@@ -22,7 +22,8 @@
 
   test('should not fetch via callback when a cache exists', () async {
     await cache.fetch(() async => 'Expensive');
-    expect(await cache.fetch(expectAsync0(() => null, count: 0)), 'Expensive');
+    expect(await cache.fetch(expectAsync0(() async => 'fake', count: 0)),
+        'Expensive');
   });
 
   test('should not fetch via callback when a future is in-flight', () async {
@@ -31,7 +32,7 @@
 
     var completer = Completer<String>();
     expect(cache.fetch(() => completer.future), completion('Expensive'));
-    expect(cache.fetch(expectAsync0(() => null, count: 0)),
+    expect(cache.fetch(expectAsync0(() async => 'fake', count: 0)),
         completion('Expensive'));
     completer.complete('Expensive');
   });
@@ -79,7 +80,10 @@
       yield '2';
       yield '3';
     }).toList();
-    expect(await cache.fetchStream(expectAsync0(() => null, count: 0)).toList(),
+    expect(
+        await cache
+            .fetchStream(expectAsync0(() => Stream.empty(), count: 0))
+            .toList(),
         ['1', '2', '3']);
   });
 
@@ -148,7 +152,7 @@
   test('should pause a cached stream without affecting others', () async {
     Stream<String> call() => Stream.fromIterable(['1', '2', '3']);
 
-    StreamSubscription sub;
+    late StreamSubscription sub;
     sub = cache.fetchStream(call).listen(expectAsync1((event) {
       if (event == '1') sub.pause();
     }));
diff --git a/test/async_memoizer_test.dart b/test/async_memoizer_test.dart
index 982f7c9..490b389 100644
--- a/test/async_memoizer_test.dart
+++ b/test/async_memoizer_test.dart
@@ -6,7 +6,7 @@
 import 'package:test/test.dart';
 
 void main() {
-  AsyncMemoizer cache;
+  late AsyncMemoizer cache;
   setUp(() => cache = AsyncMemoizer());
 
   test('runs the function only the first time runOnce() is called', () async {
diff --git a/test/cancelable_operation_test.dart b/test/cancelable_operation_test.dart
index b38c0b9..c87e43a 100644
--- a/test/cancelable_operation_test.dart
+++ b/test/cancelable_operation_test.dart
@@ -11,7 +11,7 @@
 
 void main() {
   group('without being canceled', () {
-    CancelableCompleter completer;
+    late CancelableCompleter completer;
     setUp(() {
       completer = CancelableCompleter(onCancel: expectAsync0(() {}, count: 0));
     });
@@ -116,7 +116,7 @@
 
     test('fires onCancel', () {
       var canceled = false;
-      CancelableCompleter completer;
+      late CancelableCompleter completer;
       completer = CancelableCompleter(onCancel: expectAsync0(() {
         expect(completer.isCanceled, isTrue);
         canceled = true;
@@ -243,23 +243,23 @@
   });
 
   group('then', () {
-    FutureOr<String> Function(int) onValue;
-    FutureOr<String> Function(Object, StackTrace) onError;
-    FutureOr<String> Function() onCancel;
-    bool propagateCancel;
-    CancelableCompleter<int> originalCompleter;
+    FutureOr<String> Function(int)? onValue;
+    FutureOr<String> Function(Object, StackTrace)? onError;
+    FutureOr<String> Function()? onCancel;
+    late bool propagateCancel;
+    late CancelableCompleter<int> originalCompleter;
 
     setUp(() {
       // Initialize all functions to ones that expect to not be called.
-      onValue = expectAsync1((_) => null, count: 0, id: 'onValue');
-      onError = expectAsync2((e, s) => null, count: 0, id: 'onError');
-      onCancel = expectAsync0(() => null, count: 0, id: 'onCancel');
+      onValue = expectAsync1((_) => 'Fake', count: 0, id: 'onValue');
+      onError = expectAsync2((e, s) => 'Fake', count: 0, id: 'onError');
+      onCancel = expectAsync0(() => 'Fake', count: 0, id: 'onCancel');
       propagateCancel = false;
     });
 
     CancelableOperation<String> runThen() {
       originalCompleter = CancelableCompleter();
-      return originalCompleter.operation.then(onValue,
+      return originalCompleter.operation.then(onValue!,
           onError: onError,
           onCancel: onCancel,
           propagateCancel: propagateCancel);
diff --git a/test/future_group_test.dart b/test/future_group_test.dart
index f99d06d..22e90f8 100644
--- a/test/future_group_test.dart
+++ b/test/future_group_test.dart
@@ -10,7 +10,7 @@
 import 'utils.dart';
 
 void main() {
-  FutureGroup futureGroup;
+  late FutureGroup futureGroup;
   setUp(() {
     futureGroup = FutureGroup();
   });
diff --git a/test/lazy_stream_test.dart b/test/lazy_stream_test.dart
index 2affe5b..32c6b14 100644
--- a/test/lazy_stream_test.dart
+++ b/test/lazy_stream_test.dart
@@ -10,10 +10,6 @@
 import 'utils.dart';
 
 void main() {
-  test('disallows a null callback', () {
-    expect(() => LazyStream(null), throwsArgumentError);
-  });
-
   test('calls the callback when the stream is listened', () async {
     var callbackCalled = false;
     var stream = LazyStream(expectAsync0(() {
@@ -96,7 +92,7 @@
   });
 
   test("a lazy stream can't be listened to from within its callback", () {
-    LazyStream stream;
+    late LazyStream stream;
     stream = LazyStream(expectAsync0(() {
       expect(() => stream.listen(null), throwsStateError);
       return Stream.empty();
diff --git a/test/result/result_captureAll_test.dart b/test/result/result_captureAll_test.dart
index 8e79872..b992e1f 100644
--- a/test/result/result_captureAll_test.dart
+++ b/test/result/result_captureAll_test.dart
@@ -14,7 +14,7 @@
 
 /// Helper function creating an iterable of futures.
 Iterable<Future<int>> futures(int count,
-    {bool Function(int index) throwWhen}) sync* {
+    {bool Function(int index)? throwWhen}) sync* {
   for (var i = 0; i < count; i++) {
     if (throwWhen != null && throwWhen(i)) {
       yield Future<int>.error('$i', someStack);
diff --git a/test/result/result_flattenAll_test.dart b/test/result/result_flattenAll_test.dart
index c0a8603..b87fec4 100644
--- a/test/result/result_flattenAll_test.dart
+++ b/test/result/result_flattenAll_test.dart
@@ -11,7 +11,7 @@
 
 /// Helper function creating an iterable of results.
 Iterable<Result<int>> results(int count,
-    {bool Function(int index) throwWhen}) sync* {
+    {bool Function(int index)? throwWhen}) sync* {
   for (var i = 0; i < count; i++) {
     if (throwWhen != null && throwWhen(i)) {
       yield err(i);
@@ -27,7 +27,7 @@
       expect(result, expectation);
     } else {
       expect(result.isValue, true);
-      expect(result.asValue.value, expectation.asValue.value);
+      expect(result.asValue!.value, expectation.asValue!.value);
     }
   }
 
diff --git a/test/result/result_future_test.dart b/test/result/result_future_test.dart
index 4218db2..7171005 100644
--- a/test/result/result_future_test.dart
+++ b/test/result/result_future_test.dart
@@ -9,8 +9,8 @@
 import 'package:test/test.dart';
 
 void main() {
-  Completer completer;
-  ResultFuture future;
+  late Completer completer;
+  late ResultFuture future;
   setUp(() {
     completer = Completer();
     future = ResultFuture(completer.future);
@@ -25,7 +25,7 @@
 
     // The completer calls its listeners asynchronously. We have to wait
     // before we can access the result.
-    expect(future.then((_) => future.result.asValue.value),
+    expect(future.then((_) => future.result!.asValue!.value),
         completion(equals(12)));
   });
 
@@ -36,7 +36,7 @@
     // The completer calls its listeners asynchronously. We have to wait
     // before we can access the result.
     return future.catchError((_) {}).then((_) {
-      var error = future.result.asError;
+      var error = future.result!.asError!;
       expect(error.error, equals('error'));
       expect(error.stackTrace, equals(trace));
     });
diff --git a/test/result/result_test.dart b/test/result/result_test.dart
index adbc445..154785b 100644
--- a/test/result/result_test.dart
+++ b/test/result/result_test.dart
@@ -16,7 +16,7 @@
     var result = Result<int>.value(42);
     expect(result.isValue, isTrue);
     expect(result.isError, isFalse);
-    ValueResult value = result.asValue;
+    ValueResult value = result.asValue!;
     expect(value.value, equals(42));
   });
 
@@ -24,7 +24,7 @@
     Result<int> result = ValueResult<int>(42);
     expect(result.isValue, isTrue);
     expect(result.isError, isFalse);
-    var value = result.asValue;
+    var value = result.asValue!;
     expect(value.value, equals(42));
   });
 
@@ -32,7 +32,7 @@
     var result = Result<bool>.error('BAD', stack);
     expect(result.isValue, isFalse);
     expect(result.isError, isTrue);
-    var error = result.asError;
+    var error = result.asError!;
     expect(error.error, equals('BAD'));
     expect(error.stackTrace, same(stack));
   });
@@ -50,9 +50,10 @@
     var result = Result<bool>.error('BAD');
     expect(result.isValue, isFalse);
     expect(result.isError, isTrue);
-    var error = result.asError;
+    var error = result.asError!;
     expect(error.error, equals('BAD'));
-    expect(error.stackTrace, isNull);
+    // A default stack trace is created
+    expect(error.stackTrace, isNotNull);
   });
 
   test('complete with value', () {
@@ -71,7 +72,7 @@
     var c = Completer<bool>();
     c.future.then((bool v) {
       fail('Unexpected value $v');
-    }, onError: expectAsync2((e, s) {
+    }).then<void>((_) {}, onError: expectAsync2((e, s) {
       expect(e, equals('BAD'));
       expect(s, same(stack));
     }));
@@ -108,7 +109,7 @@
     Result<bool> result = ErrorResult('BAD', stack);
     result.asFuture.then((bool v) {
       fail('Unexpected value $v');
-    }, onError: expectAsync2((e, s) {
+    }).then<void>((_) {}, onError: expectAsync2((e, s) {
       expect(e, equals('BAD'));
       expect(s, same(stack));
     }));
@@ -119,7 +120,7 @@
     Result.capture(value).then(expectAsync1((Result result) {
       expect(result.isValue, isTrue);
       expect(result.isError, isFalse);
-      var value = result.asValue;
+      var value = result.asValue!;
       expect(value.value, equals(42));
     }), onError: (e, s) {
       fail('Unexpected error: $e');
@@ -131,7 +132,7 @@
     Result.capture(value).then(expectAsync1((Result result) {
       expect(result.isValue, isFalse);
       expect(result.isError, isTrue);
-      var error = result.asError;
+      var error = result.asError!;
       expect(error.error, equals('BAD'));
       expect(error.stackTrace, same(stack));
     }), onError: (e, s) {
@@ -153,7 +154,7 @@
     var future = Future<Result<bool>>.value(Result<bool>.error('BAD', stack));
     Result.release(future).then((v) {
       fail('Unexpected value: $v');
-    }, onError: expectAsync2((e, s) {
+    }).then<void>((_) {}, onError: expectAsync2((e, s) {
       expect(e, equals('BAD'));
       expect(s, same(stack));
     }));
@@ -164,7 +165,7 @@
     var future = Future<Result<bool>>.error('BAD', stack);
     Result.release(future).then((v) {
       fail('Unexpected value: $v');
-    }, onError: expectAsync2((e, s) {
+    }).then<void>((_) {}, onError: expectAsync2((e, s) {
       expect(e, equals('BAD'));
       expect(s, same(stack));
     }));
@@ -203,15 +204,15 @@
       expect(expectedList.isEmpty, isFalse);
       Result expected = expectedList.removeFirst();
       expect(expected.isValue, isTrue);
-      expect(v, equals(expected.asValue.value));
+      expect(v, equals(expected.asValue!.value));
     }
 
     void errorListener(error, StackTrace stackTrace) {
       expect(expectedList.isEmpty, isFalse);
       Result expected = expectedList.removeFirst();
       expect(expected.isError, isTrue);
-      expect(error, equals(expected.asError.error));
-      expect(stackTrace, same(expected.asError.stackTrace));
+      expect(error, equals(expected.asError!.error));
+      expect(stackTrace, same(expected.asError!.stackTrace));
     }
 
     stream.listen(expectAsync1(dataListener, count: 2),
@@ -311,10 +312,10 @@
   expect(actual.isValue, equals(expected.isValue));
   expect(actual.isError, equals(expected.isError));
   if (actual.isValue) {
-    expect(actual.asValue.value, equals(expected.asValue.value));
+    expect(actual.asValue!.value, equals(expected.asValue!.value));
   } else {
-    expect(actual.asError.error, equals(expected.asError.error));
-    expect(actual.asError.stackTrace, same(expected.asError.stackTrace));
+    expect(actual.asError!.error, equals(expected.asError!.error));
+    expect(actual.asError!.stackTrace, same(expected.asError!.stackTrace));
   }
 }
 
@@ -334,8 +335,8 @@
   }
 
   @override
-  void addError(error, [StackTrace stack]) {
-    onError(error, stack);
+  void addError(error, [StackTrace? stack]) {
+    onError(error, stack ?? StackTrace.fromString(''));
   }
 
   @override
diff --git a/test/stream_completer_test.dart b/test/stream_completer_test.dart
index 0cd210a..ecb5621 100644
--- a/test/stream_completer_test.dart
+++ b/test/stream_completer_test.dart
@@ -77,7 +77,7 @@
     var completer = StreamCompleter<int>();
     var lastEvent = -1;
     var controller = StreamController<int>();
-    StreamSubscription subscription;
+    late StreamSubscription subscription;
     subscription = completer.stream.listen((value) {
       expect(value, lessThan(3));
       lastEvent = value;
@@ -125,7 +125,7 @@
 
   test("source stream isn't listened to until completer stream is", () async {
     var completer = StreamCompleter();
-    StreamController controller;
+    late StreamController controller;
     controller = StreamController(onListen: () {
       scheduleMicrotask(controller.close);
     });
@@ -139,13 +139,13 @@
   });
 
   test('cancelOnError true when listening before linking stream', () async {
-    var completer = StreamCompleter();
+    var completer = StreamCompleter<Object>();
     Object lastEvent = -1;
-    var controller = StreamController();
+    var controller = StreamController<Object>();
     completer.stream.listen((value) {
       expect(value, lessThan(3));
       lastEvent = value;
-    }, onError: (value) {
+    }, onError: (Object value) {
       expect(value, '3');
       lastEvent = value;
     }, onDone: unreachable('done'), cancelOnError: true);
@@ -172,9 +172,9 @@
   });
 
   test('cancelOnError true when listening after linking stream', () async {
-    var completer = StreamCompleter();
+    var completer = StreamCompleter<Object>();
     Object lastEvent = -1;
-    var controller = StreamController();
+    var controller = StreamController<Object>();
     completer.setSourceStream(controller.stream);
     controller.add(1);
     expect(controller.hasListener, isFalse);
@@ -182,7 +182,7 @@
     completer.stream.listen((value) {
       expect(value, lessThan(3));
       lastEvent = value;
-    }, onError: (value) {
+    }, onError: (Object value) {
       expect(value, '3');
       lastEvent = value;
     }, onDone: unreachable('done'), cancelOnError: true);
@@ -265,8 +265,8 @@
   });
 
   test('setting onData etc. before and after setting stream', () async {
-    var completer = StreamCompleter();
-    var controller = StreamController();
+    var completer = StreamCompleter<int>();
+    var controller = StreamController<int>();
     var subscription = completer.stream.listen(null);
     Object lastEvent = 0;
     subscription.onData((value) => lastEvent = value);
diff --git a/test/stream_group_test.dart b/test/stream_group_test.dart
index 690fbc1..eadae19 100644
--- a/test/stream_group_test.dart
+++ b/test/stream_group_test.dart
@@ -9,7 +9,7 @@
 
 void main() {
   group('single-subscription', () {
-    StreamGroup<String> streamGroup;
+    late StreamGroup<String> streamGroup;
     setUp(() {
       streamGroup = StreamGroup<String>();
     });
@@ -236,7 +236,7 @@
             StreamController<String>(onCancel: () => completer.future);
 
         var fired = false;
-        streamGroup.add(controller.stream).then((_) => fired = true);
+        streamGroup.add(controller.stream)!.then((_) => fired = true);
 
         await flushMicrotasks();
         expect(fired, isFalse);
@@ -249,7 +249,7 @@
   });
 
   group('broadcast', () {
-    StreamGroup<String> streamGroup;
+    late StreamGroup<String> streamGroup;
     setUp(() {
       streamGroup = StreamGroup<String>.broadcast();
     });
@@ -455,7 +455,7 @@
 }
 
 void regardlessOfType(StreamGroup<String> Function() newStreamGroup) {
-  StreamGroup<String> streamGroup;
+  late StreamGroup<String> streamGroup;
   setUp(() {
     streamGroup = newStreamGroup();
   });
@@ -608,7 +608,7 @@
         await flushMicrotasks();
 
         var fired = false;
-        streamGroup.remove(controller.stream).then((_) => fired = true);
+        streamGroup.remove(controller.stream)!.then((_) => fired = true);
 
         await flushMicrotasks();
         expect(fired, isFalse);
diff --git a/test/stream_queue_test.dart b/test/stream_queue_test.dart
index fa286a8..74c91e6 100644
--- a/test/stream_queue_test.dart
+++ b/test/stream_queue_test.dart
@@ -250,7 +250,7 @@
       var skip4 = events.skip(1);
       var index = 0;
       // Check that futures complete in order.
-      Func1Required<int> sequence(expectedValue, sequenceIndex) => (value) {
+      Func1Required<int?> sequence(expectedValue, sequenceIndex) => (value) {
             expect(value, expectedValue);
             expect(index, sequenceIndex);
             index++;
@@ -610,7 +610,7 @@
       var controller = StreamController<int>();
       var events = StreamQueue<int>(controller.stream);
 
-      bool hasNext;
+      bool? hasNext;
       events.hasNext.then((result) {
         hasNext = result;
       });
@@ -626,7 +626,7 @@
       var controller = StreamController<int>();
       var events = StreamQueue<int>(controller.stream);
 
-      bool hasNext;
+      bool? hasNext;
       events.hasNext.then((result) {
         hasNext = result;
       });
@@ -772,10 +772,10 @@
   });
 
   group('startTransaction operation produces a transaction that', () {
-    StreamQueue<int> events;
-    StreamQueueTransaction<int> transaction;
-    StreamQueue<int> queue1;
-    StreamQueue<int> queue2;
+    late StreamQueue<int> events;
+    late StreamQueueTransaction<int> transaction;
+    late StreamQueue<int> queue1;
+    late StreamQueue<int> queue2;
     setUp(() async {
       events = StreamQueue(createStream());
       expect(await events.next, 1);
@@ -1004,7 +1004,7 @@
   });
 
   group('withTransaction operation', () {
-    StreamQueue<int> events;
+    late StreamQueue<int> events;
     setUp(() async {
       events = StreamQueue(createStream());
       expect(await events.next, 1);
@@ -1058,7 +1058,7 @@
   });
 
   group('cancelable operation', () {
-    StreamQueue<int> events;
+    late StreamQueue<int> events;
     setUp(() async {
       events = StreamQueue(createStream());
       expect(await events.next, 1);
diff --git a/test/stream_sink_completer_test.dart b/test/stream_sink_completer_test.dart
index 69d1d8a..591f32f 100644
--- a/test/stream_sink_completer_test.dart
+++ b/test/stream_sink_completer_test.dart
@@ -10,7 +10,7 @@
 import 'utils.dart';
 
 void main() {
-  StreamSinkCompleter completer;
+  late StreamSinkCompleter completer;
   setUp(() {
     completer = StreamSinkCompleter();
   });
@@ -21,10 +21,10 @@
       completer.setDestinationSink(sink);
       completer.sink..add(1)..add(2)..add(3)..add(4);
 
-      expect(sink.results[0].asValue.value, equals(1));
-      expect(sink.results[1].asValue.value, equals(2));
-      expect(sink.results[2].asValue.value, equals(3));
-      expect(sink.results[3].asValue.value, equals(4));
+      expect(sink.results[0].asValue!.value, equals(1));
+      expect(sink.results[1].asValue!.value, equals(2));
+      expect(sink.results[2].asValue!.value, equals(3));
+      expect(sink.results[3].asValue!.value, equals(4));
     });
 
     test('error events are forwarded', () {
@@ -32,8 +32,8 @@
       completer.setDestinationSink(sink);
       completer.sink..addError('oh no')..addError("that's bad");
 
-      expect(sink.results[0].asError.error, equals('oh no'));
-      expect(sink.results[1].asError.error, equals("that's bad"));
+      expect(sink.results[0].asError!.error, equals('oh no'));
+      expect(sink.results[1].asError!.error, equals("that's bad"));
     });
 
     test('addStream is forwarded', () async {
@@ -49,10 +49,10 @@
       controller.addError("that's bad");
       await flushMicrotasks();
 
-      expect(sink.results[0].asValue.value, equals(1));
-      expect(sink.results[1].asError.error, equals('oh no'));
-      expect(sink.results[2].asValue.value, equals(2));
-      expect(sink.results[3].asError.error, equals("that's bad"));
+      expect(sink.results[0].asValue!.value, equals(1));
+      expect(sink.results[1].asError!.error, equals('oh no'));
+      expect(sink.results[2].asValue!.value, equals(2));
+      expect(sink.results[3].asError!.error, equals("that's bad"));
       expect(sink.isClosed, isFalse);
 
       controller.close();
@@ -121,10 +121,10 @@
       completer.setDestinationSink(sink);
       await flushMicrotasks();
 
-      expect(sink.results[0].asValue.value, equals(1));
-      expect(sink.results[1].asValue.value, equals(2));
-      expect(sink.results[2].asValue.value, equals(3));
-      expect(sink.results[3].asValue.value, equals(4));
+      expect(sink.results[0].asValue!.value, equals(1));
+      expect(sink.results[1].asValue!.value, equals(2));
+      expect(sink.results[2].asValue!.value, equals(3));
+      expect(sink.results[3].asValue!.value, equals(4));
     });
 
     test('error events are forwarded', () async {
@@ -135,8 +135,8 @@
       completer.setDestinationSink(sink);
       await flushMicrotasks();
 
-      expect(sink.results[0].asError.error, equals('oh no'));
-      expect(sink.results[1].asError.error, equals("that's bad"));
+      expect(sink.results[0].asError!.error, equals('oh no'));
+      expect(sink.results[1].asError!.error, equals("that's bad"));
     });
 
     test('addStream is forwarded', () async {
@@ -154,10 +154,10 @@
       completer.setDestinationSink(sink);
       await flushMicrotasks();
 
-      expect(sink.results[0].asValue.value, equals(1));
-      expect(sink.results[1].asError.error, equals('oh no'));
-      expect(sink.results[2].asValue.value, equals(2));
-      expect(sink.results[3].asError.error, equals("that's bad"));
+      expect(sink.results[0].asValue!.value, equals(1));
+      expect(sink.results[1].asError!.error, equals('oh no'));
+      expect(sink.results[2].asValue!.value, equals(2));
+      expect(sink.results[3].asError!.error, equals("that's bad"));
       expect(sink.isClosed, isFalse);
     });
 
@@ -258,9 +258,9 @@
       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));
+      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 {
diff --git a/test/stream_sink_transformer_test.dart b/test/stream_sink_transformer_test.dart
index 8493971..e5f6baa 100644
--- a/test/stream_sink_transformer_test.dart
+++ b/test/stream_sink_transformer_test.dart
@@ -10,7 +10,7 @@
 import 'utils.dart';
 
 void main() {
-  StreamController controller;
+  late StreamController controller;
   setUp(() {
     controller = StreamController();
   });
@@ -18,7 +18,7 @@
   group('fromStreamTransformer', () {
     test('transforms data events', () {
       var transformer = StreamSinkTransformer.fromStreamTransformer(
-          StreamTransformer.fromHandlers(handleData: (i, sink) {
+          StreamTransformer.fromHandlers(handleData: (int i, sink) {
         sink.add(i * 2);
       }));
       var sink = transformer.bind(controller.sink);
@@ -117,7 +117,7 @@
   group('fromHandlers', () {
     test('transforms data events', () {
       var transformer =
-          StreamSinkTransformer.fromHandlers(handleData: (i, sink) {
+          StreamSinkTransformer.fromHandlers(handleData: (int i, sink) {
         sink.add(i * 2);
       });
       var sink = transformer.bind(controller.sink);
diff --git a/test/stream_splitter_test.dart b/test/stream_splitter_test.dart
index 3748c5c..3493193 100644
--- a/test/stream_splitter_test.dart
+++ b/test/stream_splitter_test.dart
@@ -8,8 +8,8 @@
 import 'package:test/test.dart';
 
 void main() {
-  StreamController<int> controller;
-  StreamSplitter splitter;
+  late StreamController<int> controller;
+  late StreamSplitter splitter;
   setUp(() {
     controller = StreamController<int>();
     splitter = StreamSplitter<int>(controller.stream);
diff --git a/test/stream_zip_test.dart b/test/stream_zip_test.dart
index 19c3dec..84948bf 100644
--- a/test/stream_zip_test.dart
+++ b/test/stream_zip_test.dart
@@ -9,7 +9,7 @@
 
 /// Create an error with the same values as [base], except that it throwsA
 /// when seeing the value [errorValue].
-Stream streamError(Stream base, int errorValue, error) {
+Stream streamError(Stream base, int errorValue, Object error) {
   return base.map((x) => (x == errorValue) ? throw error : x);
 }
 
@@ -306,7 +306,7 @@
     var s2 = Stream.fromIterable([1, 3, 5, 7]);
     var sz = StreamZip([s1, s2]);
     var ctr = 0;
-    StreamSubscription sub;
+    late StreamSubscription sub;
     sub = sz.listen(expectAsync1((v) {
       expect(v, equals([ctr * 2, ctr * 2 + 1]));
       if (ctr == 1) {
diff --git a/test/stream_zip_zone_test.dart b/test/stream_zip_zone_test.dart
index 68cd723..50af6b6 100644
--- a/test/stream_zip_zone_test.dart
+++ b/test/stream_zip_zone_test.dart
@@ -33,7 +33,7 @@
     var outer = Zone.current;
     runZoned(() {
       var newZone1 = Zone.current;
-      StreamSubscription sub;
+      late StreamSubscription sub;
       sub = stream.listen(expectAsync1((v) {
         expect(v, 42);
         expect(Zone.current, newZone1);
diff --git a/test/subscription_stream_test.dart b/test/subscription_stream_test.dart
index 98a809f..8be0572 100644
--- a/test/subscription_stream_test.dart
+++ b/test/subscription_stream_test.dart
@@ -22,7 +22,7 @@
     var stream = createStream();
     var skips = 0;
     var completer = Completer();
-    StreamSubscription<int> subscription;
+    late StreamSubscription<int> subscription;
     subscription = stream.listen((value) {
       ++skips;
       expect(value, skips);
@@ -72,8 +72,9 @@
   group('cancelOnError source:', () {
     for (var sourceCancels in [false, true]) {
       group('${sourceCancels ? "yes" : "no"}:', () {
-        SubscriptionStream subscriptionStream;
-        Future onCancel; // Completes if source stream is canceled before done.
+        late SubscriptionStream subscriptionStream;
+        late Future
+            onCancel; // Completes if source stream is canceled before done.
         setUp(() {
           var cancelCompleter = Completer();
           var source = createErrorStream(cancelCompleter);
@@ -159,7 +160,7 @@
   yield 4;
 }
 
-Stream<int> createErrorStream([Completer onCancel]) async* {
+Stream<int> createErrorStream([Completer? onCancel]) async* {
   var canceled = true;
   try {
     yield 1;
diff --git a/test/subscription_transformer_test.dart b/test/subscription_transformer_test.dart
index f0e3a70..8278ea0 100644
--- a/test/subscription_transformer_test.dart
+++ b/test/subscription_transformer_test.dart
@@ -90,7 +90,7 @@
           subscriptionTransformer(handleCancel: expectAsync1((inner) {
         callbackInvoked = true;
         inner.cancel();
-        return null;
+        return Future.value();
       }))).listen(expectAsync1((_) {}, count: 0));
 
       await flushMicrotasks();
@@ -248,11 +248,12 @@
   });
 
   group('when the outer subscription is canceled but the inner is not', () {
-    StreamSubscription subscription;
+    late StreamSubscription subscription;
     setUp(() {
       var controller = StreamController();
       subscription = controller.stream
-          .transform(subscriptionTransformer(handleCancel: (_) => null))
+          .transform(
+              subscriptionTransformer(handleCancel: (_) => Future.value()))
           .listen(expectAsync1((_) {}, count: 0),
               onError: expectAsync2((_, __) {}, count: 0),
               onDone: expectAsync0(() {}, count: 0));
diff --git a/test/typed_wrapper/stream_subscription_test.dart b/test/typed_wrapper/stream_subscription_test.dart
index 2a11546..33d28a8 100644
--- a/test/typed_wrapper/stream_subscription_test.dart
+++ b/test/typed_wrapper/stream_subscription_test.dart
@@ -11,10 +11,11 @@
 
 void main() {
   group('with valid types, forwards', () {
-    StreamController controller;
-    StreamSubscription wrapper;
-    bool isCanceled;
+    late StreamController controller;
+    late StreamSubscription wrapper;
+    late bool isCanceled;
     setUp(() {
+      isCanceled = false;
       controller = StreamController<Object>(onCancel: () {
         isCanceled = true;
       });
@@ -67,10 +68,11 @@
   });
 
   group('with invalid types,', () {
-    StreamController controller;
-    StreamSubscription wrapper;
-    bool isCanceled;
+    late StreamController controller;
+    late StreamSubscription wrapper;
+    late bool isCanceled;
     setUp(() {
+      isCanceled = false;
       controller = StreamController<Object>(onCancel: () {
         isCanceled = true;
       });
diff --git a/test/utils.dart b/test/utils.dart
index a99fe9b..4cc743e 100644
--- a/test/utils.dart
+++ b/test/utils.dart
@@ -25,7 +25,7 @@
 Matcher throwsZoned(matcher) => predicate((void Function() callback) {
       var firstError = true;
       runZoned(callback,
-          onError: expectAsync2((error, StackTrace stackTrace) {
+          onError: expectAsync2((Object error, StackTrace stackTrace) {
             if (firstError) {
               expect(error, matcher);
               firstError = false;
@@ -37,11 +37,11 @@
     });
 
 /// A matcher that runs a callback in its own zone and asserts that that zone
-/// emits a [CastError].
-final throwsZonedCastError = throwsZoned(TypeMatcher<CastError>());
+/// emits a [TypeError].
+final throwsZonedCastError = throwsZoned(TypeMatcher<TypeError>());
 
-/// A matcher that matches a callback or future that throws a [CastError].
-final throwsCastError = throwsA(TypeMatcher<CastError>());
+/// A matcher that matches a callback or future that throws a [TypeError].
+final throwsCastError = throwsA(TypeMatcher<TypeError>());
 
 /// A badly behaved stream which throws if it's ever listened to.
 ///
@@ -67,7 +67,7 @@
   @override
   void add(T event) {}
   @override
-  void addError(error, [StackTrace stackTrace]) {}
+  void addError(error, [StackTrace? stackTrace]) {}
   @override
   Future addStream(Stream<T> stream) async {}
   @override
@@ -95,7 +95,7 @@
   ///
   /// If [onDone] is passed, it's called when the user calls [close]. Its result
   /// is piped to the [done] future.
-  TestSink({void Function() onDone}) : _onDone = onDone ?? (() {});
+  TestSink({void Function()? onDone}) : _onDone = onDone ?? (() {});
 
   @override
   void add(T event) {
@@ -103,7 +103,7 @@
   }
 
   @override
-  void addError(error, [StackTrace stackTrace]) {
+  void addError(error, [StackTrace? stackTrace]) {
     results.add(Result<T>.error(error, stackTrace));
   }