Merge pull request #5 from hoc081098/v2.1.0

2.1.0

Author: hoc081098

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 2.1.0 - Aug 28, 2020
+
+-   State stream returned from `RxReduxStore` will not replay the latest state
+    (Use `RxReduxStore.state` getter instead).
+
 ## 2.0.0 - Aug 27, 2020
 
 -   Added `Logger` which allows logging current state, action and new state.

lib/src/redux_store_stream_transformer.dart

@@ -6,6 +6,8 @@ import 'logger.dart';
 import 'reducer.dart';
 import 'reducer_exception.dart';
 import 'side_effect.dart';
+import 'utils.dart';
+import 'wrapper_action.dart';
 
 /// Redux store stream Extension for Stream of actions.
 extension ReduxStoreExt<Action> on Stream<Action> {
@@ -94,7 +96,7 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
   Stream<S> bind(Stream<A> stream) {
     StreamController<S> controller;
     List<StreamSubscription<dynamic>> subscriptions;
-    StreamController<_WrapperAction<A>> actionController;
+    StreamController<WrapperAction<A>> actionController;
 
     void onListen() {
       S state;
@@ -107,13 +109,13 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
         return;
       }
 
-      void onDataActually(_WrapperAction<A> wrapper) {
+      void onDataActually(WrapperAction<A> wrapper) {
         final action = wrapper.action;
         final type = wrapper.type;
         final currentState = state;
 
         // add initial state
-        if (type == _ActionType.initial) {
+        if (identical(type, ActionType.initial)) {
           final message = '\n'
               '  ⟶ Action       : $type\n'
               '  ⟹ Current state: $currentState';
@@ -149,18 +151,18 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
         }
       }
 
-      actionController = StreamController<_WrapperAction<A>>.broadcast();
+      actionController = StreamController<WrapperAction<A>>.broadcast();
 
       // Call reducer on each action.
       final subscriptionActionController =
           actionController.stream.listen(onDataActually);
 
       // Add initial action
-      actionController.add(_WrapperAction(null, _ActionType.initial));
+      actionController.add(WrapperAction(null, ActionType.initial));
 
       // Listening to upstream actions
       final subscriptionUpstream = stream
-          .map((action) => _WrapperAction(action, _ActionType.external))
+          .map((action) => WrapperAction(action, ActionType.external))
           .listen(
             actionController.add,
             onError: controller.addError,
@@ -207,7 +209,7 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
   }
 
   Iterable<StreamSubscription<dynamic>> _listenSideEffects(
-    StreamController<_WrapperAction<A>> actionController,
+    StreamController<WrapperAction<A>> actionController,
     GetState<S> getState,
     StreamController<S> controller,
   ) {
@@ -224,8 +226,8 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
         }
 
         return actions
-            .map((action) =>
-                _WrapperAction(action, _ActionType.sideEffect(index)))
+            .map(
+                (action) => WrapperAction(action, ActionType.sideEffect(index)))
             .listen(
               actionController.add,
               onError: controller.addError,
@@ -236,59 +238,3 @@ class ReduxStoreStreamTransformer<A, S> extends StreamTransformerBase<A, S> {
     );
   }
 }
-
-//
-// Internal
-//
-
-@sealed
-abstract class _ActionType {
-  const _ActionType.empty();
-
-  static const initial = _Initial();
-  static const external = _External();
-
-  const factory _ActionType.sideEffect(int index) = _SideEffect;
-
-  @override
-  String toString() {
-    if (this is _Initial) {
-      return '⭍';
-    }
-    if (this is _External) {
-      return '↓';
-    }
-    if (this is _SideEffect) {
-      return '⟳${(this as _SideEffect).index}';
-    }
-    throw StateError('Unknown $this');
-  }
-}
-
-class _Initial extends _ActionType {
-  const _Initial() : super.empty();
-}
-
-class _External extends _ActionType {
-  const _External() : super.empty();
-}
-
-class _SideEffect extends _ActionType {
-  final int index;
-
-  const _SideEffect(this.index) : super.empty();
-}
-
-class _WrapperAction<A> {
-  final A action;
-  final _ActionType type;
-
-  _WrapperAction(this.action, this.type);
-}
-
-extension _MapIndexedIterableExtensison<T> on Iterable<T> {
-  Iterable<R> mapIndexed<R>(R Function(int, T) mapper) {
-    var index = 0;
-    return map((t) => mapper(index++, t));
-  }
-}

lib/src/store.dart

@@ -31,7 +31,7 @@ class RxReduxStore<A, S> {
   final void Function(A) _dispatch;
 
   final GetState<S> _getState;
-  final Stream<S> Function() _stateStream;
+  final Stream<S> _stateStream;
   final Stream<A> _actionStream;
 
   final Future<void> Function() _dispose;
@@ -62,18 +62,22 @@ class RxReduxStore<A, S> {
     bool Function(S previous, S next) equals,
     RxReduxLogger logger,
   }) {
-    final actionSubject = StreamController<A>.broadcast(sync: true);
+    final actionController = StreamController<A>(sync: true);
     final actionOutputController = StreamController<A>.broadcast(sync: true);
 
-    final stateStream = actionSubject.stream.reduxStore<S>(
-      initialStateSupplier: () => initialState,
-      sideEffects: [
-        ...sideEffects,
-        _onEachActionSideEffect(actionOutputController),
-      ],
-      reducer: reducer,
-      logger: logger,
-    );
+    final stateStream = actionController.stream
+        .reduxStore<S>(
+          initialStateSupplier: () => initialState,
+          sideEffects: [
+            ...sideEffects,
+            _onEachActionSideEffect(actionOutputController),
+          ],
+          reducer: reducer,
+          logger: logger,
+        )
+        .distinct(equals)
+        .skip(1)
+        .asBroadcastStream(onCancel: (subscription) => subscription.cancel());
 
     var currentState = initialState;
     final subscription = stateStream.listen(
@@ -82,18 +86,14 @@ class RxReduxStore<A, S> {
     );
 
     return RxReduxStore._(
-      actionSubject.add,
+      actionController.add,
       () => currentState,
-      () => () async* {
-        yield currentState;
-        yield* stateStream;
-      }()
-          .distinct(equals),
+      stateStream,
       actionOutputController.stream,
-      () => Future.wait([
-        actionSubject.close(),
-        subscription.cancel(),
-      ]),
+      () async {
+        await actionController.close();
+        await subscription.cancel();
+      },
     );
   }
 
@@ -115,7 +115,7 @@ class RxReduxStore<A, S> {
   ///         return LoginWidget(state); // build widget based on state.
   ///       },
   ///     );
-  Stream<S> get stateStream => _stateStream();
+  Stream<S> get stateStream => _stateStream;
 
   /// Get current state synchronously.
   /// This is useful for filling `initialData` when using `StreamBuilder` in Flutter.

lib/src/utils.dart

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

lib/src/wrapper_action.dart

@@ -0,0 +1,47 @@
+import 'package:meta/meta.dart';
+
+// ignore_for_file: public_member_api_docs, missing_return
+
+@sealed
+abstract class ActionType {
+  const ActionType.empty();
+
+  static final initial = _Initial();
+  static final external = _External();
+
+  factory ActionType.sideEffect(int index) = _SideEffect;
+
+  @override
+  String toString() {
+    if (this is _Initial) {
+      return '⭍';
+    }
+    if (this is _External) {
+      return '↓';
+    }
+    if (this is _SideEffect) {
+      return '⟳${(this as _SideEffect).index}';
+    }
+  }
+}
+
+class _Initial extends ActionType {
+  _Initial() : super.empty();
+}
+
+class _External extends ActionType {
+  _External() : super.empty();
+}
+
+class _SideEffect extends ActionType {
+  final int index;
+
+  _SideEffect(this.index) : super.empty();
+}
+
+class WrapperAction<A> {
+  final A action;
+  final ActionType type;
+
+  WrapperAction(this.action, this.type);
+}

test/store_test.dart

@@ -46,14 +46,15 @@ void main() {
       expect(store.state, 0);
     });
 
-    test('Get state stream that emits initial state', () {
+    test('Get state stream that never emits', () {
       final store = RxReduxStore<int, int>(
         initialState: 0,
         sideEffects: [],
         reducer: (s, a) => s + a,
       );
 
-      expect(store.stateStream, emits(0));
+      store.stateStream.listen((event) => expect(true, isFalse));
+      Future<void>.delayed(const Duration(seconds: 1));
     });
 
     test('Get state stream', () async {
@@ -67,7 +68,6 @@ void main() {
         store.stateStream,
         emitsInOrder(
           <String>[
-            '0',
             '0+1',
             '0+1+2',
             '0+1+2+3',
@@ -82,7 +82,8 @@ void main() {
       store.dispatch(3);
 
       await future;
-      expect(store.stateStream, emits('0+1+2+3'));
+      store.stateStream.listen((_) => expect(true, isFalse));
+      await Future<void>.delayed(const Duration(seconds: 1));
     });
 
     test('Get state stream with SideEffects', () async {
@@ -123,7 +124,6 @@ void main() {
         store.stateStream,
         emitsInOrder(
           <int>[
-            0,
             1,
             3,
             6,
@@ -146,7 +146,6 @@ void main() {
         store.stateStream,
         emitsInOrder(
           <int>[
-            6,
             7,
             9,
             12,
@@ -286,16 +285,10 @@ void main() {
 
       await rxReduxStore.dispose();
 
-      rxReduxStore.stateStream.listen(
-        expectAsync1(
-          (v) => expect(v, '0'),
-          count: 1,
-        ),
-        onDone: expectAsync0(
-          () {},
-          count: 1,
-        ),
-      );
+      expect(rxReduxStore.state, '0');
+      expect(() => rxReduxStore.dispatch(0), throwsStateError);
+      expect(rxReduxStore.actionStream, emitsDone);
+      expect(rxReduxStore.stateStream, emitsDone);
     });
 
     test('Error handler called when reducer throws', () async {