Null safety migration

analysis_options.yaml

@@ -7,6 +7,8 @@ analyzer:
     todo: ignore
     # Lint provided by pkg:pedantic – should fix this!
     unawaited_futures: ignore
+  enable-experiment:
+    - non-nullable
 
 linter:
   rules:

lib/src/async_cache.dart

@@ -29,13 +29,13 @@ class AsyncCache<T> {
   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.
   ///
@@ -78,12 +78,12 @@ class AsyncCache<T> {
     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.

lib/src/cancelable_operation.dart

@@ -35,7 +35,7 @@ class CancelableOperation<T> {
   /// 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 @@ class CancelableOperation<T> {
     value.then((value) {
       controller.add(value);
       controller.close();
-    }, onError: (error, StackTrace stackTrace) {
+    }, onError: (Object error, StackTrace stackTrace) {
       controller.addError(error, stackTrace);
       controller.close();
     });
@@ -68,7 +68,10 @@ class CancelableOperation<T> {
   /// 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]) {
+  ///
+  /// If [T] is not nullable then [cancellationValue] must provide a valid
+  /// value, otherwise it will throw on cancellation.
+  Future<T> valueOrCancellation([T? cancellationValue]) {
     var completer = Completer<T>.sync();
     value.then((result) => completer.complete(result),
         onError: completer.completeError);
@@ -93,8 +96,8 @@ class CancelableOperation<T> {
   /// 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);
@@ -109,7 +112,7 @@ class CancelableOperation<T> {
           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 +148,7 @@ class CancelableCompleter<T> {
   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 +158,14 @@ class CancelableCompleter<T> {
   ///
   /// [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 +182,7 @@ class CancelableCompleter<T> {
   ///
   /// 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 +202,14 @@ class CancelableCompleter<T> {
     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 +223,8 @@ class CancelableCompleter<T> {
 
     return _cancelMemo.runOnce(() {
       _isCanceled = true;
-      if (_onCancel != null) return _onCancel();
+      var onCancel = _onCancel;
+      if (onCancel != null) return onCancel();
     });
   }
 }

lib/src/delegate/event_sink.dart

@@ -33,7 +33,7 @@ class DelegatingEventSink<T> implements EventSink<T> {
   }
 
   @override
-  void addError(error, [StackTrace stackTrace]) {
+  void addError(error, [StackTrace? stackTrace]) {
     _sink.addError(error, stackTrace);
   }
 

lib/src/delegate/future.dart

@@ -25,18 +25,18 @@ class DelegatingFuture<T> implements Future<T> {
   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
   Future<T> whenComplete(FutureOr Function() action) =>
       _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);
 }

lib/src/delegate/stream_sink.dart

@@ -36,7 +36,7 @@ class DelegatingStreamSink<T> implements StreamSink<T> {
   }
 
   @override
-  void addError(error, [StackTrace stackTrace]) {
+  void addError(error, [StackTrace? stackTrace]) {
     _sink.addError(error, stackTrace);
   }
 

lib/src/delegate/stream_subscription.dart

@@ -31,22 +31,22 @@ class DelegatingStreamSubscription<T> implements StreamSubscription<T> {
           : 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 @@ class DelegatingStreamSubscription<T> implements StreamSubscription<T> {
   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;

lib/src/future_group.dart

@@ -44,13 +44,13 @@ class FutureGroup<T> implements Sink<Future<T>> {
   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 @@ class FutureGroup<T> implements Sink<Future<T>> {
       _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 @@ class FutureGroup<T> implements Sink<Future<T>> {
     _closed = true;
     if (_pending != 0) return;
     if (_completer.isCompleted) return;
-    _completer.complete(_values);
+    _completer.complete(_values.whereType<T>().toList());
   }
 }

lib/src/lazy_stream.dart

@@ -15,7 +15,7 @@ import 'utils.dart';
 /// 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 @@ class LazyStream<T> extends Stream<T> {
   }
 
   @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();
 

lib/src/null_stream_sink.dart

@@ -43,12 +43,12 @@ class NullStreamSink<T> implements StreamSink<T> {
   ///
   /// 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 @@ class NullStreamSink<T> implements StreamSink<T> {
   }
 
   @override
-  void addError(error, [StackTrace stackTrace]) {
+  void addError(Object error, [StackTrace? stackTrace]) {
     _checkEventAllowed();
   }
 
@@ -70,7 +70,7 @@ class NullStreamSink<T> implements StreamSink<T> {
     _checkEventAllowed();
 
     _addingStream = true;
-    var future = stream.listen(null).cancel() ?? Future.value();
+    var future = stream.listen(null).cancel();
     return future.whenComplete(() {
       _addingStream = false;
     });

lib/src/result/capture_sink.dart

@@ -18,7 +18,7 @@ class CaptureSink<T> implements EventSink<T> {
   }
 
   @override
-  void addError(Object error, [StackTrace stackTrace]) {
+  void addError(Object error, [StackTrace? stackTrace]) {
     _sink.add(Result.error(error, stackTrace));
   }
 

lib/src/result/error.dart

@@ -8,7 +8,7 @@ import 'result.dart';
 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 @@ class ErrorResult implements Result<Null> {
   @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 @@ class ErrorResult implements Result<Null> {
   }
 
   @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.
   ///

lib/src/result/future.dart

@@ -16,8 +16,8 @@ class ResultFuture<T> extends DelegatingFuture<T> {
   /// 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) {

lib/src/result/release_sink.dart

@@ -18,7 +18,7 @@ class ReleaseSink<T> implements EventSink<Result<T>> {
   }
 
   @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);