null_safety!

Author: hoc081098

lib/src/reducer_exception.dart

@@ -1,5 +1,3 @@
-import 'package:meta/meta.dart';
-
 /// Exception thrown by reducer
 class ReducerException<A, S> implements Exception {
   /// Action passed into reducer
@@ -16,10 +14,10 @@ class ReducerException<A, S> implements Exception {
 
   /// Construct a [ReducerException]
   const ReducerException({
-    @required this.action,
-    @required this.state,
-    @required this.error,
-    this.stackTrace,
+    required this.action,
+    required this.state,
+    required this.error,
+    required this.stackTrace,
   });
 
   @override

lib/src/redux_store_stream_transformer.dart

@@ -1,7 +1,5 @@
 import 'dart:async';
 
-import 'package:meta/meta.dart';
-
 import 'logger.dart';
 import 'reducer.dart';
 import 'reducer_exception.dart';
@@ -34,10 +32,10 @@ extension ReduxStoreExt<Action> on Stream<Action> {
   /// * Param [State] The type of the State
   /// * Param [Action] The type of the Actions
   Stream<State> reduxStore<State>({
-    @required State Function() initialStateSupplier,
-    @required Iterable<SideEffect<Action, State>> sideEffects,
-    @required Reducer<Action, State> reducer,
-    RxReduxLogger logger,
+    required State Function() initialStateSupplier,
+    required Iterable<SideEffect<Action, State>> sideEffects,
+    required Reducer<Action, State> reducer,
+    RxReduxLogger? logger,
   }) =>
       transform(
         ReduxStoreStreamTransformer(
@@ -66,7 +64,7 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
   final S Function() _initialStateSupplier;
   final Iterable<SideEffect<A, S>> _sideEffects;
   final Reducer<A, S> _reducer;
-  final RxReduxLogger _logger;
+  final RxReduxLogger? _logger;
 
   /// * Param [initialStateSupplier]  A function that computes the initial state. The computation is
   /// * done lazily once an observer subscribes. The computed initial state will be emitted directly
@@ -77,26 +75,20 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
   /// * Param [S] The type of the State
   /// * Param [A] The type of the Actions
   ReduxStoreStreamTransformer({
-    @required S Function() initialStateSupplier,
-    @required Iterable<SideEffect<A, S>> sideEffects,
-    @required Reducer<A, S> reducer,
-    RxReduxLogger logger,
-  })  : assert(initialStateSupplier != null,
-            'initialStateSupplier cannot be null'),
-        assert(sideEffects != null, 'sideEffects cannot be null'),
-        assert(sideEffects.every((sideEffect) => sideEffect != null),
-            'All sideEffects must be not null'),
-        assert(reducer != null, 'reducer cannot be null'),
-        _initialStateSupplier = initialStateSupplier,
+    required S Function() initialStateSupplier,
+    required Iterable<SideEffect<A, S>> sideEffects,
+    required Reducer<A, S> reducer,
+    RxReduxLogger? logger,
+  })  : _initialStateSupplier = initialStateSupplier,
         _sideEffects = sideEffects,
         _reducer = reducer,
         _logger = logger;
 
   @override
   Stream<S> bind(Stream<A> stream) {
-    StreamController<S> controller;
-    List<StreamSubscription<dynamic>> subscriptions;
-    StreamController<WrapperAction<A>> actionController;
+    late StreamController<S> controller;
+    List<StreamSubscription<dynamic>>? subscriptions;
+    StreamController<WrapperAction<A>>? _actionController;
 
     void onListen() {
       S state;
@@ -110,7 +102,6 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
       }
 
       void onDataActually(WrapperAction<A> wrapper) {
-        final action = wrapper.action;
         final type = wrapper.type;
         final currentState = state;
 
@@ -123,6 +114,7 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
           return controller.add(currentState);
         }
 
+        final action = wrapper.action!;
         try {
           final newState = _reducer(currentState, action);
           controller.add(newState);
@@ -151,7 +143,8 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
         }
       }
 
-      actionController = StreamController<WrapperAction<A>>.broadcast();
+      final actionController =
+          _actionController = StreamController<WrapperAction<A>>.broadcast();
 
       // Call reducer on each action.
       final subscriptionActionController =
@@ -170,16 +163,24 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
           );
 
       final getState = () => state;
+      final actionStream = actionController.stream
+          .map((wrapper) => wrapper.action!)
+          .asBroadcastStream(onCancel: (s) => s.cancel());
 
       subscriptions = [
-        ..._listenSideEffects(actionController, getState, controller),
+        ..._listenSideEffects(
+          actionController,
+          getState,
+          controller,
+          actionStream,
+        ),
         subscriptionUpstream,
         subscriptionActionController
       ];
     }
 
     Future<void> onCancel() async {
-      final future = actionController?.close();
+      final future = _actionController?.close();
       final cancelFutures = subscriptions?.map((s) => s.cancel());
       final futures = [...?cancelFutures, if (future != null) future];
 
@@ -199,8 +200,8 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
       controller = StreamController<S>(
         sync: true,
         onListen: onListen,
-        onPause: () => subscriptions.forEach((s) => s.pause()),
-        onResume: () => subscriptions.forEach((s) => s.resume()),
+        onPause: () => subscriptions?.forEach((s) => s.pause()),
+        onResume: () => subscriptions?.forEach((s) => s.resume()),
         onCancel: onCancel,
       );
     }
@@ -211,26 +212,24 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
   Iterable<StreamSubscription<dynamic>> _listenSideEffects(
     StreamController<WrapperAction<A>> actionController,
     GetState<S> getState,
-    StreamController<S> controller,
+    StreamController<S> stateController,
+    Stream<A> actionStream,
   ) {
     return _sideEffects.mapIndexed(
       (index, sideEffect) {
         Stream<A> actions;
         try {
-          actions = sideEffect(
-            actionController.stream.map((wrapper) => wrapper.action),
-            getState,
-          );
+          actions = sideEffect(actionStream, getState);
         } catch (e, s) {
           actions = Stream.error(e, s);
         }
 
+        final sideEffectAction = ActionType.sideEffect(index);
         return actions
-            .map(
-                (action) => WrapperAction(action, ActionType.sideEffect(index)))
+            .map((action) => WrapperAction(action, sideEffectAction))
             .listen(
               actionController.add,
-              onError: controller.addError,
+              onError: stateController.addError,
               // Swallow onDone because just if one SideEffect reaches onDone
               // we don't want to make everything incl. ReduxStore and other SideEffects reach onDone
             );

lib/src/store.dart

@@ -1,7 +1,5 @@
 import 'dart:async';
 
-import 'package:meta/meta.dart';
-
 import 'logger.dart';
 import 'reducer.dart';
 import 'redux_store_stream_transformer.dart';
@@ -12,7 +10,7 @@ import 'side_effect.dart';
 SideEffect<A, S> _onEachActionSideEffect<A, S>(StreamSink<A> outputSink) {
   return (actions, _) {
     final neverController = StreamController<A>();
-    StreamSubscription<A> subscription;
+    late StreamSubscription<A> subscription;
 
     neverController.onListen =
         () => subscription = actions.listen(outputSink.add);
@@ -57,12 +55,12 @@ class RxReduxStore<A, S> {
   /// If calling it returns true, the new state will not be emitted.
   /// If [equals] is omitted, the '==' operator is used.
   factory RxReduxStore({
-    @required S initialState,
-    @required List<SideEffect<A, S>> sideEffects,
-    @required Reducer<A, S> reducer,
-    void Function(Object, StackTrace) errorHandler,
-    bool Function(S previous, S next) equals,
-    RxReduxLogger logger,
+    required S initialState,
+    required List<SideEffect<A, S>> sideEffects,
+    required Reducer<A, S> reducer,
+    void Function(Object, StackTrace)? errorHandler,
+    bool Function(S previous, S next)? equals,
+    RxReduxLogger? logger,
   }) {
     final actionController = StreamController<A>(sync: true);
     final actionOutputController = StreamController<A>.broadcast(sync: true);
@@ -82,7 +80,7 @@ class RxReduxStore<A, S> {
         .asBroadcastStream(onCancel: (subscription) => subscription.cancel());
 
     var currentState = initialState;
-    final subscriptions = <StreamSubscription<Object>>[
+    final subscriptions = <StreamSubscription<dynamic>>[
       stateStream.listen(
         (newState) => currentState = newState,
         onError: errorHandler,

lib/src/utils.dart

@@ -1,8 +1,10 @@
 // ignore_for_file: public_member_api_docs
 
 extension MapIndexedIterableExtensison<T> on Iterable<T> {
-  Iterable<R> mapIndexed<R>(R Function(int, T) mapper) {
+  Iterable<R> mapIndexed<R>(R Function(int, T) mapper) sync* {
     var index = 0;
-    return map((t) => mapper(index++, t));
+    for (final t in this) {
+      yield mapper(index++, t);
+    }
   }
 }

lib/src/wrapper_action.dart

@@ -1,6 +1,6 @@
 import 'package:meta/meta.dart';
 
-// ignore_for_file: public_member_api_docs, missing_return
+// ignore_for_file: public_member_api_docs
 
 @sealed
 abstract class ActionType {
@@ -14,14 +14,15 @@ abstract class ActionType {
   @override
   String toString() {
     if (this is _Initial) {
-      return '⭍';
+      return '↯';
     }
     if (this is _External) {
       return '↓';
     }
     if (this is _SideEffect) {
       return '⟳${(this as _SideEffect).index}';
     }
+    throw StateError('What is $this');
   }
 }
 
@@ -40,7 +41,7 @@ class _SideEffect extends ActionType {
 }
 
 class WrapperAction<A> {
-  final A action;
+  final A? action;
   final ActionType type;
 
   WrapperAction(this.action, this.type);

pubspec.yaml

@@ -7,15 +7,11 @@ repository: https://github.com/hoc081098/rx_redux.git
 issue_tracker: https://github.com/hoc081098/rx_redux/issues
 
 environment:
-  sdk: '>=2.7.0 <3.0.0'
+  sdk: '>=2.12.0-0 <3.0.0'
 
 dependencies:
-  meta: ^1.1.8
 
 dev_dependencies:
-  pedantic: ^1.9.0
-  test: ^1.15.3
-  rxdart:
-    git:
-      url: "https://github.com/ReactiveX/rxdart.git"
-      ref: "master"
+  pedantic: ^1.10.0-nullsafety.3
+  test: ^1.16.0-nullsafety.12
+  rxdart: ^0.26.0-nullsafety.0