Feature/update re listener modifier

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 1.1.0
+* Updated `ReListenerModifier` approach
+
 ## 1.0.1
 * Added `guardState` method to `ReState` to allow safe emitting of state changes without try-catch blocks
 

example/lib/main.dart

@@ -98,12 +98,11 @@ class ReCounterStateActionEvent
   ReCounterStateActionEvent() : super(0) {
     on<IncrementCounter>(
       (event) => _increment(),
+      modifier: ReListenerModifiers.throttleTime(const Duration(seconds: 1)),
     );
     on<ResetCounter>(
       (event) => _reset(),
-      modifier: (eventFlow) => eventFlow.debounceTime(
-        const Duration(seconds: 1),
-      ),
+      modifier: ReListenerModifiers.debounceTime(const Duration(seconds: 1)),
     );
 
     listenState(

example/pubspec.lock

@@ -37,10 +37,10 @@ packages:
     dependency: transitive
     description:
       name: collection
-      sha256: "4a07be6cb69c84d677a6c3096fcf960cc3285a8330b4603e0d463d15d9bd934c"
+      sha256: f092b211a4319e98e5ff58223576de6c2803db36221657b46c82574721240687
       url: "https://pub.dev"
     source: hosted
-    version: "1.17.1"
+    version: "1.17.2"
   cupertino_icons:
     dependency: "direct main"
     description:
@@ -66,47 +66,39 @@ packages:
     dependency: "direct dev"
     description:
       name: flutter_lints
-      sha256: aeb0b80a8b3709709c9cc496cdc027c5b3216796bc0af0ce1007eaf24464fd4c
+      sha256: "2118df84ef0c3ca93f96123a616ae8540879991b8b57af2f81b76a7ada49b2a4"
       url: "https://pub.dev"
     source: hosted
-    version: "2.0.1"
+    version: "2.0.2"
   flutter_test:
     dependency: "direct dev"
     description: flutter
     source: sdk
     version: "0.0.0"
-  js:
-    dependency: transitive
-    description:
-      name: js
-      sha256: f2c445dce49627136094980615a031419f7f3eb393237e4ecd97ac15dea343f3
-      url: "https://pub.dev"
-    source: hosted
-    version: "0.6.7"
   lints:
     dependency: transitive
     description:
       name: lints
-      sha256: "5e4a9cd06d447758280a8ac2405101e0e2094d2a1dbdd3756aec3fe7775ba593"
+      sha256: "0a217c6c989d21039f1498c3ed9f3ed71b354e69873f13a8dfc3c9fe76f1b452"
       url: "https://pub.dev"
     source: hosted
-    version: "2.0.1"
+    version: "2.1.1"
   matcher:
     dependency: transitive
     description:
       name: matcher
-      sha256: "6501fbd55da300384b768785b83e5ce66991266cec21af89ab9ae7f5ce1c4cbb"
+      sha256: "1803e76e6653768d64ed8ff2e1e67bea3ad4b923eb5c56a295c3e634bad5960e"
       url: "https://pub.dev"
     source: hosted
-    version: "0.12.15"
+    version: "0.12.16"
   material_color_utilities:
     dependency: transitive
     description:
       name: material_color_utilities
-      sha256: d92141dc6fe1dad30722f9aa826c7fbc896d021d792f80678280601aff8cf724
+      sha256: "9528f2f296073ff54cb9fee677df673ace1218163c3bc7628093e7eed5203d41"
       url: "https://pub.dev"
     source: hosted
-    version: "0.2.0"
+    version: "0.5.0"
   meta:
     dependency: transitive
     description:
@@ -147,10 +139,10 @@ packages:
     dependency: transitive
     description:
       name: source_span
-      sha256: dd904f795d4b4f3b870833847c461801f6750a9fa8e61ea5ac53f9422b31f250
+      sha256: "53e943d4206a5e30df338fd4c6e7a077e02254531b138a15aec3bd143c1a8b3c"
       url: "https://pub.dev"
     source: hosted
-    version: "1.9.1"
+    version: "1.10.0"
   stack_trace:
     dependency: transitive
     description:
@@ -187,10 +179,10 @@ packages:
     dependency: transitive
     description:
       name: test_api
-      sha256: eb6ac1540b26de412b3403a163d919ba86f6a973fe6cc50ae3541b80092fdcfb
+      sha256: "75760ffd7786fffdfb9597c35c5b27eaeec82be8edfb6d71d32651128ed7aab8"
       url: "https://pub.dev"
     source: hosted
-    version: "0.5.1"
+    version: "0.6.0"
   vector_math:
     dependency: transitive
     description:
@@ -199,6 +191,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "2.1.4"
+  web:
+    dependency: transitive
+    description:
+      name: web
+      sha256: dc8ccd225a2005c1be616fe02951e2e342092edf968cf0844220383757ef8f10
+      url: "https://pub.dev"
+    source: hosted
+    version: "0.1.4-beta"
 sdks:
-  dart: ">=3.0.0-0 <4.0.0"
+  dart: ">=3.1.0-185.0.dev <4.0.0"
   flutter: ">=1.17.0"

lib/re_listener_modifiers.dart

@@ -1,3 +1,42 @@
 library re_listener_modifiers;
 
+import 'package:re_state_action/src/typedefs/re_types.dart';
+import 'package:rxdart/transformers.dart';
+
 export 'package:rxdart/transformers.dart';
+
+/// A collection of static methods that return [ReListenerModifier] functions.
+/// These functions can be used to modify the behavior of a
+/// [ReListenerModifier].
+abstract class ReListenerModifiers {
+  /// Returns a [ReListenerModifier] that applies a mapper function to the
+  /// listener and returns the flattened result.
+  static ReListenerModifier<T> flatMap<T>() =>
+      (listener, mapper) => listener.flatMap(mapper);
+
+  /// Returns a [ReListenerModifier] that applies a mapper function to the
+  /// listener and returns the result of the latest mapped stream.
+  static ReListenerModifier<T> switchMap<T>() =>
+      (listener, mapper) => listener.switchMap(mapper);
+
+  /// Returns a [ReListenerModifier] that applies a mapper function to the
+  /// listener and ignores all events until the mapper completes.
+  static ReListenerModifier<T> exhaustMap<T>() =>
+      (listener, mapper) => listener.exhaustMap(mapper);
+
+  /// Returns a [ReListenerModifier] that applies a mapper function to the
+  /// listener and returns the result of the mapped stream as soon as it is
+  /// available.
+  static ReListenerModifier<T> asyncExpand<T>() =>
+      (listener, mapper) => listener.asyncExpand(mapper);
+
+  /// Returns a [ReListenerModifier] that applies a [Duration] [duration] to
+  /// the listener and debounces the events.
+  static ReListenerModifier<T> debounceTime<T>(Duration duration) =>
+      (listener, mapper) => listener.debounceTime(duration).flatMap(mapper);
+
+  /// Returns a [ReListenerModifier] that applies a [Duration] [duration] to
+  ///  the listener and throttles the events.
+  static ReListenerModifier<T> throttleTime<T>(Duration duration) =>
+      (listener, mapper) => listener.throttleTime(duration).flatMap(mapper);
+}

lib/src/mixins/re_action_mixin.dart

@@ -4,6 +4,7 @@ import 'dart:async';
 
 import 'package:flutter/foundation.dart';
 import 'package:re_state_action/src/typedefs/re_types.dart';
+import 'package:re_state_action/src/utils/re_listener_utils.dart';
 import 'package:re_state_action/src/utils/re_subscription_holder.dart';
 import 'package:rxdart/rxdart.dart';
 
@@ -51,18 +52,10 @@ mixin ReActionMixin<Action> on ReSubscriptionHolder {
   /// [listener] is called whenever an [Action] is emitted.
   ///
   /// [modifier] is used to modify the stream of actions before it is listened to.
-  ///
-  /// [onError] is called whenever an error occurs.
-  ///
-  /// [onDone] is called when the stream is closed.
-  ///
-  /// [cancelOnError] is used to cancel the subscription when an error occurs.
+
   void listenAction(
     ReActionCallback<Action> listener, {
     ReListenerModifier<Action>? modifier,
-    Function? onError,
-    void Function()? onDone,
-    bool cancelOnError = false,
   }) {
     if (!_isInitialized) {
       throw StateError(
@@ -78,15 +71,11 @@ mixin ReActionMixin<Action> on ReSubscriptionHolder {
       );
     }
 
-    final listenerModifier = modifier ?? (listener) => listener;
+    final listenerModifier = modifier ?? reListenerModifier();
+    final listenerMapper = reListenerMapper(listener);
 
     final subscription = subscriptions.add(
-      listenerModifier(actionStream).listen(
-        listener,
-        onError: onError,
-        onDone: onDone,
-        cancelOnError: cancelOnError,
-      ),
+      listenerModifier(actionStream, listenerMapper).listen(null),
     );
     _actionSubscriptions[listener] = subscription;
   }

lib/src/mixins/re_event_mixin.dart

@@ -1,5 +1,6 @@
 import 'package:flutter/foundation.dart';
 import 'package:re_state_action/re_state_action.dart';
+import 'package:re_state_action/src/utils/re_listener_utils.dart';
 import 'package:re_state_action/src/utils/re_subscription_holder.dart';
 import 'package:rxdart/rxdart.dart';
 
@@ -24,19 +25,10 @@ mixin ReEventMixin<Event> on ReSubscriptionHolder {
   /// [callback] is called whenever an event of type [T] is dispatched.
   ///
   /// [modifier] is used to modify the stream of events before it is listened.
-  ///
-  /// [onError] is called whenever an error occurs.
-  ///
-  /// [onDone] is called when the stream is closed.
-  ///
-  /// [cancelOnError] is used to cancel the subscription when an error occurs.
   @protected
   void on<T extends Event>(
     ReEventCallback<T> callback, {
     ReListenerModifier<T>? modifier,
-    Function? onError,
-    void Function()? onDone,
-    bool cancelOnError = false,
   }) {
     if (!_isInitialized) {
       throw StateError(
@@ -62,15 +54,11 @@ mixin ReEventMixin<Event> on ReSubscriptionHolder {
     _eventsMap[type] = callback;
 
     final stream = _eventsNotifier.stream.whereType<T>();
-    final listenerModifier = modifier ?? (stream) => stream;
+    final listenerModifier = modifier ?? reListenerModifier();
+    final listenerMapper = reListenerMapper(callback);
 
     final subscription = subscriptions.add(
-      listenerModifier(stream).listen(
-        callback,
-        onError: onError,
-        onDone: onDone,
-        cancelOnError: cancelOnError,
-      ),
+      listenerModifier(stream, listenerMapper).listen(null),
     );
 
     subscriptions.add(subscription);

lib/src/mixins/re_state_mixin.dart

@@ -4,6 +4,7 @@ import 'dart:async';
 
 import 'package:flutter/foundation.dart';
 import 'package:re_state_action/src/typedefs/re_types.dart';
+import 'package:re_state_action/src/utils/re_listener_utils.dart';
 import 'package:re_state_action/src/utils/re_subscription_holder.dart';
 import 'package:rxdart/rxdart.dart';
 
@@ -90,18 +91,9 @@ mixin ReStateMixin<State> on ReSubscriptionHolder {
   /// [listener] is called whenever a [State] is emitted.
   ///
   /// [modifier] is used to modify the stream of state before it is listened to.
-  ///
-  /// [onError] is called whenever an error occurs.
-  ///
-  /// [onDone] is called when the stream is closed.
-  ///
-  /// [cancelOnError] is used to cancel the subscription when an error occurs.
   void listenState(
     ReStateCallback<State> listener, {
     ReListenerModifier<State>? modifier,
-    Function? onError,
-    void Function()? onDone,
-    bool cancelOnError = false,
   }) {
     if (!_isInitialized) {
       throw StateError(
@@ -117,15 +109,11 @@ mixin ReStateMixin<State> on ReSubscriptionHolder {
       );
     }
 
-    final listenerModifier = modifier ?? (listener) => listener;
+    final listenerModifier = modifier ?? reListenerModifier();
+    final listenerMapper = reListenerMapper(listener);
 
     final subscription = subscriptions.add(
-      listenerModifier(stateStream).listen(
-        listener,
-        onError: onError,
-        onDone: onDone,
-        cancelOnError: cancelOnError,
-      ),
+      listenerModifier(stateStream, listenerMapper).listen(null),
     );
 
     _stateSubscriptions[listener] = subscription;

lib/src/typedefs/re_types.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:flutter/widgets.dart';
 
 /// A function that builds a widget given the current [State] and the [child].
@@ -8,12 +10,12 @@ typedef ReStateBuilder<State> = Widget Function(
 );
 
 /// A function that is called when the [State] changes.
-typedef ReStateCallback<State> = void Function(
+typedef ReStateCallback<State> = FutureOr<void> Function(
   State state,
 );
 
 /// A function that is called when an [Action] is dispatched.
-typedef ReActionCallback<Action> = void Function(
+typedef ReActionCallback<Action> = FutureOr<void> Function(
   Action action,
 );
 
@@ -31,12 +33,16 @@ typedef ReActionListenerCondition<Action> = bool Function(
   Action currentAction,
 );
 
+/// A function that maps the [T] to a [Stream] of [T].
+typedef ReListenerMapper<T> = Stream<T> Function(T callback);
+
 /// A function that modifies the listener to be called when the [T] changes.
 typedef ReListenerModifier<T> = Stream<T> Function(
   Stream<T> listener,
+  ReListenerMapper<T> mapper,
 );
 
 /// A function that is called when an [Event] is dispatched.
-typedef ReEventCallback<Event> = void Function(
+typedef ReEventCallback<Event> = FutureOr<void> Function(
   Event event,
 );

lib/src/utils/re_listener_utils.dart

@@ -0,0 +1,26 @@
+import 'dart:async';
+
+import 'package:re_state_action/re_listener_modifiers.dart';
+import 'package:re_state_action/src/typedefs/re_types.dart';
+
+/// @nodoc
+ReListenerMapper<T> reListenerMapper<T>(FutureOr<void> Function(T) callback) =>
+    (data) {
+      final controller = StreamController<T>.broadcast(sync: true);
+
+      Future<void> handleData() async {
+        try {
+          await callback(data);
+        } catch (_) {
+          rethrow;
+        } finally {
+          await controller.done;
+        }
+      }
+
+      handleData();
+      return controller.stream;
+    };
+
+/// @nodoc
+ReListenerModifier<T> reListenerModifier<T>() => ReListenerModifiers.flatMap();

pubspec.yaml

@@ -2,7 +2,7 @@ name: re_state_action
 description: A simple state management library for Flutter based on States and Actions managed by Streams with RxDart.
 repository: https://github.com/alvarobcprado/re_state_action
 issue_tracker: https://github.com/alvarobcprado/re_state_action/issues
-version: 1.0.1
+version: 1.1.0
 
 environment:
   sdk: ">=2.12.0 <4.0.0"