new change to use intaleq_map sdk 04-16-4

packages/get/lib/get_rx/src/rx_stream/get_stream.dart

@@ -0,0 +1,209 @@
+part of rx_stream;
+
+/// [GetStream] is the lightest and most performative way of working
+/// with events at Dart. You sintaxe is like StreamController, but it works
+/// with simple callbacks. In this way, every event calls only one function.
+/// There is no buffering, to very low memory consumption.
+/// event [add] will add a object to stream. [addError] will add a error
+/// to stream. [listen] is a very light StreamSubscription interface.
+/// Is possible take the last value with [value] property.
+class GetStream<T> {
+  void Function()? onListen;
+  void Function()? onPause;
+  void Function()? onResume;
+  FutureOr<void> Function()? onCancel;
+
+  GetStream({this.onListen, this.onPause, this.onResume, this.onCancel});
+  List<LightSubscription<T>>? _onData = <LightSubscription<T>>[];
+
+  bool? _isBusy = false;
+
+  FutureOr<bool?> removeSubscription(LightSubscription<T> subs) async {
+    if (!_isBusy!) {
+      return _onData!.remove(subs);
+    } else {
+      await Future.delayed(Duration.zero);
+      return _onData?.remove(subs);
+    }
+  }
+
+  FutureOr<void> addSubscription(LightSubscription<T> subs) async {
+    if (!_isBusy!) {
+      return _onData!.add(subs);
+    } else {
+      await Future.delayed(Duration.zero);
+      return _onData!.add(subs);
+    }
+  }
+
+  int? get length => _onData?.length;
+
+  bool get hasListeners => _onData!.isNotEmpty;
+
+  void _notifyData(T data) {
+    _isBusy = true;
+    for (final item in _onData!) {
+      if (!item.isPaused) {
+        item._data?.call(data);
+      }
+    }
+    _isBusy = false;
+  }
+
+  void _notifyError(Object error, [StackTrace? stackTrace]) {
+    assert(!isClosed, 'You cannot add errors to a closed stream.');
+    _isBusy = true;
+    var itemsToRemove = <LightSubscription<T>>[];
+    for (final item in _onData!) {
+      if (!item.isPaused) {
+        if (stackTrace != null) {
+          item._onError?.call(error, stackTrace);
+        } else {
+          item._onError?.call(error);
+        }
+
+        if (item.cancelOnError ?? false) {
+          //item.cancel?.call();
+          itemsToRemove.add(item);
+          item.pause();
+          item._onDone?.call();
+        }
+      }
+    }
+    for (final item in itemsToRemove) {
+      _onData!.remove(item);
+    }
+    _isBusy = false;
+  }
+
+  void _notifyDone() {
+    assert(!isClosed, 'You cannot close a closed stream.');
+    _isBusy = true;
+    for (final item in _onData!) {
+      if (!item.isPaused) {
+        item._onDone?.call();
+      }
+    }
+    _isBusy = false;
+  }
+
+  T? _value;
+
+  T? get value => _value;
+
+  void add(T event) {
+    assert(!isClosed, 'You cannot add event to closed Stream');
+    _value = event;
+    _notifyData(event);
+  }
+
+  bool get isClosed => _onData == null;
+
+  void addError(Object error, [StackTrace? stackTrace]) {
+    assert(!isClosed, 'You cannot add error to closed Stream');
+    _notifyError(error, stackTrace);
+  }
+
+  void close() {
+    assert(!isClosed, 'You cannot close a closed Stream');
+    _notifyDone();
+    _onData = null;
+    _isBusy = null;
+    _value = null;
+  }
+
+  LightSubscription<T> listen(void Function(T event) onData,
+      {Function? onError, void Function()? onDone, bool? cancelOnError}) {
+    final subs = LightSubscription<T>(
+      removeSubscription,
+      onPause: onPause,
+      onResume: onResume,
+      onCancel: onCancel,
+    )
+      ..onData(onData)
+      ..onError(onError)
+      ..onDone(onDone)
+      ..cancelOnError = cancelOnError;
+    addSubscription(subs);
+    onListen?.call();
+    return subs;
+  }
+
+  Stream<T> get stream =>
+      GetStreamTransformation(addSubscription, removeSubscription);
+}
+
+class LightSubscription<T> extends StreamSubscription<T> {
+  final RemoveSubscription<T> _removeSubscription;
+  LightSubscription(this._removeSubscription,
+      {this.onPause, this.onResume, this.onCancel});
+  final void Function()? onPause;
+  final void Function()? onResume;
+  final FutureOr<void> Function()? onCancel;
+
+  bool? cancelOnError = false;
+
+  @override
+  Future<void> cancel() {
+    _removeSubscription(this);
+    onCancel?.call();
+    return Future.value();
+  }
+
+  OnData<T>? _data;
+
+  Function? _onError;
+
+  Callback? _onDone;
+
+  bool _isPaused = false;
+
+  @override
+  void onData(OnData<T>? handleData) => _data = handleData;
+
+  @override
+  void onError(Function? handleError) => _onError = handleError;
+
+  @override
+  void onDone(Callback? handleDone) => _onDone = handleDone;
+
+  @override
+  void pause([Future<void>? resumeSignal]) {
+    _isPaused = true;
+    onPause?.call();
+  }
+
+  @override
+  void resume() {
+    _isPaused = false;
+    onResume?.call();
+  }
+
+  @override
+  bool get isPaused => _isPaused;
+
+  @override
+  Future<E> asFuture<E>([E? futureValue]) => Future.value(futureValue);
+}
+
+class GetStreamTransformation<T> extends Stream<T> {
+  final AddSubscription<T> _addSubscription;
+  final RemoveSubscription<T> _removeSubscription;
+  GetStreamTransformation(this._addSubscription, this._removeSubscription);
+
+  @override
+  LightSubscription<T> listen(void Function(T event)? onData,
+      {Function? onError, void Function()? onDone, bool? cancelOnError}) {
+    final subs = LightSubscription<T>(_removeSubscription)
+      ..onData(onData)
+      ..onError(onError)
+      ..onDone(onDone);
+    _addSubscription(subs);
+    return subs;
+  }
+}
+
+typedef RemoveSubscription<T> = FutureOr<bool?> Function(
+    LightSubscription<T> subs);
+
+typedef AddSubscription<T> = FutureOr<void> Function(LightSubscription<T> subs);

packages/get/lib/get_rx/src/rx_stream/mini_stream.dart

@@ -0,0 +1,203 @@
+part of rx_stream;
+
+class Node<T> {
+  T? data;
+  Node<T>? next;
+  Node({this.data, this.next});
+}
+
+class MiniSubscription<T> {
+  const MiniSubscription(
+      this.data, this.onError, this.onDone, this.cancelOnError, this.listener);
+  final OnData<T> data;
+  final Function? onError;
+  final Callback? onDone;
+  final bool cancelOnError;
+
+  Future<void> cancel() async => listener.removeListener(this);
+
+  final FastList<T> listener;
+}
+
+class MiniStream<T> {
+  FastList<T> listenable = FastList<T>();
+
+  late T _value;
+
+  T get value => _value;
+
+  set value(T val) {
+    add(val);
+  }
+
+  void add(T event) {
+    _value = event;
+    listenable._notifyData(event);
+  }
+
+  void addError(Object error, [StackTrace? stackTrace]) {
+    listenable._notifyError(error, stackTrace);
+  }
+
+  int get length => listenable.length;
+
+  bool get hasListeners => listenable.isNotEmpty;
+
+  bool get isClosed => _isClosed;
+
+  MiniSubscription<T> listen(void Function(T event) onData,
+      {Function? onError,
+      void Function()? onDone,
+      bool cancelOnError = false}) {
+    final subs = MiniSubscription<T>(
+      onData,
+      onError,
+      onDone,
+      cancelOnError,
+      listenable,
+    );
+    listenable.addListener(subs);
+    return subs;
+  }
+
+  bool _isClosed = false;
+
+  void close() {
+    if (_isClosed) {
+      throw 'You can not close a closed Stream';
+    }
+    listenable._notifyDone();
+    listenable.clear();
+    _isClosed = true;
+  }
+}
+
+class FastList<T> {
+  Node<MiniSubscription<T>>? _head;
+
+  void _notifyData(T data) {
+    var currentNode = _head;
+    do {
+      currentNode?.data?.data(data);
+      currentNode = currentNode?.next;
+    } while (currentNode != null);
+  }
+
+  void _notifyDone() {
+    var currentNode = _head;
+    do {
+      currentNode?.data?.onDone?.call();
+      currentNode = currentNode?.next;
+    } while (currentNode != null);
+  }
+
+  void _notifyError(Object error, [StackTrace? stackTrace]) {
+    var currentNode = _head;
+    while (currentNode != null) {
+      currentNode.data!.onError?.call(error, stackTrace);
+      currentNode = currentNode.next;
+    }
+  }
+
+  /// Checks if this list is empty
+  bool get isEmpty => _head == null;
+
+  bool get isNotEmpty => !isEmpty;
+
+  /// Returns the length of this list
+  int get length {
+    var length = 0;
+    var currentNode = _head;
+
+    while (currentNode != null) {
+      currentNode = currentNode.next;
+      length++;
+    }
+    return length;
+  }
+
+  /// Shows the element at position [position]. `null` for invalid positions.
+  MiniSubscription<T>? _elementAt(int position) {
+    if (isEmpty || length < position || position < 0) return null;
+
+    var node = _head;
+    var current = 0;
+
+    while (current != position) {
+      node = node!.next;
+      current++;
+    }
+    return node!.data;
+  }
+
+  /// Inserts [data] at the end of the list.
+  void addListener(MiniSubscription<T> data) {
+    var newNode = Node(data: data);
+
+    if (isEmpty) {
+      _head = newNode;
+    } else {
+      var currentNode = _head!;
+      while (currentNode.next != null) {
+        currentNode = currentNode.next!;
+      }
+      currentNode.next = newNode;
+    }
+  }
+
+  bool contains(T element) {
+    var length = this.length;
+    for (var i = 0; i < length; i++) {
+      if (_elementAt(i) == element) return true;
+      if (length != this.length) {
+        throw ConcurrentModificationError(this);
+      }
+    }
+    return false;
+  }
+
+  void removeListener(MiniSubscription<T> element) {
+    var length = this.length;
+    for (var i = 0; i < length; i++) {
+      if (_elementAt(i) == element) {
+        _removeAt(i);
+        break;
+      }
+    }
+  }
+
+  void clear() {
+    var length = this.length;
+    for (var i = 0; i < length; i++) {
+      _removeAt(i);
+    }
+  }
+
+  MiniSubscription<T>? _removeAt(int position) {
+    var index = 0;
+    var currentNode = _head;
+    Node<MiniSubscription<T>>? previousNode;
+
+    if (isEmpty || length < position || position < 0) {
+      throw Exception('Invalid position');
+    } else if (position == 0) {
+      _head = _head!.next;
+    } else {
+      while (index != position) {
+        previousNode = currentNode;
+        currentNode = currentNode!.next;
+        index++;
+      }
+
+      if (previousNode == null) {
+        _head = null;
+      } else {
+        previousNode.next = currentNode!.next;
+      }
+
+      currentNode!.next = null;
+    }
+
+    return currentNode!.data;
+  }
+}

packages/get/lib/get_rx/src/rx_stream/rx_stream.dart

@@ -0,0 +1,8 @@
+library rx_stream;
+
+import 'dart:async';
+
+import '../rx_typedefs/rx_typedefs.dart';
+
+part 'get_stream.dart';
+part 'mini_stream.dart';

packages/get/lib/get_rx/src/rx_typedefs/rx_typedefs.dart

@@ -0,0 +1,3 @@
+typedef Condition = bool Function();
+typedef OnData<T> = void Function(T data);
+typedef Callback = void Function();

packages/get/lib/get_rx/src/rx_types/rx_core/rx_impl.dart

@@ -0,0 +1,394 @@
+part of rx_types;
+
+/// global object that registers against `GetX` and `Obx`, and allows the
+/// reactivity
+/// of those `Widgets` and Rx values.
+
+mixin RxObjectMixin<T> on NotifyManager<T> {
+  late T _value;
+
+  /// Makes a direct update of [value] adding it to the Stream
+  /// useful when you make use of Rx for custom Types to referesh your UI.
+  ///
+  /// Sample:
+  /// ```
+  ///  class Person {
+  ///     String name, last;
+  ///     int age;
+  ///     Person({this.name, this.last, this.age});
+  ///     @override
+  ///     String toString() => '$name $last, $age years old';
+  ///  }
+  ///
+  /// final person = Person(name: 'John', last: 'Doe', age: 18).obs;
+  /// person.value.name = 'Roi';
+  /// person.refresh();
+  /// print( person );
+  /// ```
+  void refresh() {
+    subject.add(value);
+  }
+
+  /// updates the value to `null` and adds it to the Stream.
+  /// Even with null-safety coming, is still an important feature to support, as
+  /// `call()` doesn't accept `null` values. For instance,
+  /// `InputDecoration.errorText` has to be null to not show the "error state".
+  ///
+  /// Sample:
+  /// ```
+  /// final inputError = ''.obs..nil();
+  /// print('${inputError.runtimeType}: $inputError'); // outputs > RxString: null
+  /// ```
+  // void nil() {
+  //   subject.add(_value = null);
+  // }
+
+  /// Makes this Rx looks like a function so you can update a new
+  /// value using `rx(someOtherValue)`. Practical to assign the Rx directly
+  /// to some Widget that has a signature ::onChange( value )
+  ///
+  /// Example:
+  /// ```
+  /// final myText = 'GetX rocks!'.obs;
+  ///
+  /// // in your Constructor, just to check it works :P
+  /// ever( myText, print ) ;
+  ///
+  /// // in your build(BuildContext) {
+  /// TextField(
+  ///   onChanged: myText,
+  /// ),
+  ///```
+  T call([T? v]) {
+    if (v != null) {
+      value = v;
+    }
+    return value;
+  }
+
+  bool firstRebuild = true;
+  bool sentToStream = false;
+
+  /// Same as `toString()` but using a getter.
+  String get string => value.toString();
+
+  @override
+  String toString() => value.toString();
+
+  /// Returns the json representation of `value`.
+  dynamic toJson() => value;
+
+  /// This equality override works for _RxImpl instances and the internal
+  /// values.
+  @override
+  // ignore: avoid_equals_and_hash_code_on_mutable_classes
+  bool operator ==(Object o) {
+    // Todo, find a common implementation for the hashCode of different Types.
+    if (o is T) return value == o;
+    if (o is RxObjectMixin<T>) return value == o.value;
+    return false;
+  }
+
+  @override
+  // ignore: avoid_equals_and_hash_code_on_mutable_classes
+  int get hashCode => _value.hashCode;
+
+  /// Updates the [value] and adds it to the stream, updating the observer
+  /// Widget, only if it's different from the previous value.
+  set value(T val) {
+    if (subject.isClosed) return;
+    sentToStream = false;
+    if (_value == val && !firstRebuild) return;
+    firstRebuild = false;
+    _value = val;
+    sentToStream = true;
+    subject.add(_value);
+  }
+
+  /// Returns the current [value]
+  T get value {
+    RxInterface.proxy?.addListener(subject);
+    return _value;
+  }
+
+  Stream<T> get stream => subject.stream;
+
+  /// Returns a [StreamSubscription] similar to [listen], but with the
+  /// added benefit that it primes the stream with the current [value], rather
+  /// than waiting for the next [value]. This should not be called in [onInit]
+  /// or anywhere else during the build process.
+  StreamSubscription<T> listenAndPump(void Function(T event) onData,
+      {Function? onError, void Function()? onDone, bool? cancelOnError}) {
+    final subscription = listen(
+      onData,
+      onError: onError,
+      onDone: onDone,
+      cancelOnError: cancelOnError,
+    );
+
+    subject.add(value);
+
+    return subscription;
+  }
+
+  /// Binds an existing `Stream<T>` to this Rx<T> to keep the values in sync.
+  /// You can bind multiple sources to update the value.
+  /// Closing the subscription will happen automatically when the observer
+  /// Widget (`GetX` or `Obx`) gets unmounted from the Widget tree.
+  void bindStream(Stream<T> stream) {
+    final listSubscriptions =
+        _subscriptions[subject] ??= <StreamSubscription>[];
+    listSubscriptions.add(stream.listen((va) => value = va));
+  }
+}
+
+class RxNotifier<T> = RxInterface<T> with NotifyManager<T>;
+
+mixin NotifyManager<T> {
+  GetStream<T> subject = GetStream<T>();
+  final _subscriptions = <GetStream, List<StreamSubscription>>{};
+
+  bool get canUpdate => _subscriptions.isNotEmpty;
+
+  /// This is an internal method.
+  /// Subscribe to changes on the inner stream.
+  void addListener(GetStream<T> rxGetx) {
+    if (!_subscriptions.containsKey(rxGetx)) {
+      final subs = rxGetx.listen((data) {
+        if (!subject.isClosed) subject.add(data);
+      });
+      final listSubscriptions =
+          _subscriptions[rxGetx] ??= <StreamSubscription>[];
+      listSubscriptions.add(subs);
+    }
+  }
+
+  StreamSubscription<T> listen(
+    void Function(T) onData, {
+    Function? onError,
+    void Function()? onDone,
+    bool? cancelOnError,
+  }) =>
+      subject.listen(
+        onData,
+        onError: onError,
+        onDone: onDone,
+        cancelOnError: cancelOnError ?? false,
+      );
+
+  /// Closes the subscriptions for this Rx, releasing the resources.
+  void close() {
+    _subscriptions.forEach((getStream, _subscriptions) {
+      for (final subscription in _subscriptions) {
+        subscription.cancel();
+      }
+    });
+
+    _subscriptions.clear();
+    subject.close();
+  }
+}
+
+/// Base Rx class that manages all the stream logic for any Type.
+abstract class _RxImpl<T> extends RxNotifier<T> with RxObjectMixin<T> {
+  _RxImpl(T initial) {
+    _value = initial;
+  }
+
+  void addError(Object error, [StackTrace? stackTrace]) {
+    subject.addError(error, stackTrace);
+  }
+
+  Stream<R> map<R>(R mapper(T? data)) => stream.map(mapper);
+
+  /// Uses a callback to update [value] internally, similar to [refresh],
+  /// but provides the current value as the argument.
+  /// Makes sense for custom Rx types (like Models).
+  ///
+  /// Sample:
+  /// ```
+  ///  class Person {
+  ///     String name, last;
+  ///     int age;
+  ///     Person({this.name, this.last, this.age});
+  ///     @override
+  ///     String toString() => '$name $last, $age years old';
+  ///  }
+  ///
+  /// final person = Person(name: 'John', last: 'Doe', age: 18).obs;
+  /// person.update((person) {
+  ///   person.name = 'Roi';
+  /// });
+  /// print( person );
+  /// ```
+  void update(void fn(T? val)) {
+    fn(_value);
+    subject.add(_value);
+  }
+
+  /// Following certain practices on Rx data, we might want to react to certain
+  /// listeners when a value has been provided, even if the value is the same.
+  /// At the moment, we ignore part of the process if we `.call(value)` with
+  /// the same value since it holds the value and there's no real
+  /// need triggering the entire process for the same value inside, but
+  /// there are other situations where we might be interested in
+  /// triggering this.
+  ///
+  /// For example, supposed we have a `int seconds = 2` and we want to animate
+  /// from invisible to visible a widget in two seconds:
+  /// RxEvent<int>.call(seconds);
+  /// then after a click happens, you want to call a RxEvent<int>.call(seconds).
+  /// By doing `call(seconds)`, if the value being held is the same,
+  /// the listeners won't trigger, hence we need this new `trigger` function.
+  /// This will refresh the listener of an AnimatedWidget and will keep
+  /// the value if the Rx is kept in memory.
+  /// Sample:
+  /// ```
+  /// Rx<Int> secondsRx = RxInt();
+  /// secondsRx.listen((value) => print("$value seconds set"));
+  ///
+  /// secondsRx.call(2);      // This won't trigger any listener, since the value is the same
+  /// secondsRx.trigger(2);   // This will trigger the listener independently from the value.
+  /// ```
+  ///
+  void trigger(T v) {
+    var firstRebuild = this.firstRebuild;
+    value = v;
+    // If it's not the first rebuild, the listeners have been called already
+    // So we won't call them again.
+    if (!firstRebuild && !sentToStream) {
+      subject.add(v);
+    }
+  }
+}
+
+class RxBool extends Rx<bool> {
+  RxBool(bool initial) : super(initial);
+  @override
+  String toString() {
+    return value ? "true" : "false";
+  }
+}
+
+class RxnBool extends Rx<bool?> {
+  RxnBool([bool? initial]) : super(initial);
+  @override
+  String toString() {
+    return "$value";
+  }
+}
+
+extension RxBoolExt on Rx<bool> {
+  bool get isTrue => value;
+
+  bool get isFalse => !isTrue;
+
+  bool operator &(bool other) => other && value;
+
+  bool operator |(bool other) => other || value;
+
+  bool operator ^(bool other) => !other == value;
+
+  /// Toggles the bool [value] between false and true.
+  /// A shortcut for `flag.value = !flag.value;`
+  /// FIXME: why return this? fluent interface is not
+  ///  not really a dart thing since we have '..' operator
+  // ignore: avoid_returning_this
+  Rx<bool> toggle() {
+    subject.add(_value = !_value);
+    return this;
+  }
+}
+
+extension RxnBoolExt on Rx<bool?> {
+  bool? get isTrue => value;
+
+  bool? get isFalse {
+    if (value != null) return !isTrue!;
+    return null;
+  }
+
+  bool? operator &(bool other) {
+    if (value != null) {
+      return other && value!;
+    }
+    return null;
+  }
+
+  bool? operator |(bool other) {
+    if (value != null) {
+      return other || value!;
+    }
+    return null;
+  }
+
+  bool? operator ^(bool other) => !other == value;
+
+  /// Toggles the bool [value] between false and true.
+  /// A shortcut for `flag.value = !flag.value;`
+  /// FIXME: why return this? fluent interface is not
+  ///  not really a dart thing since we have '..' operator
+  // ignore: avoid_returning_this
+  Rx<bool?>? toggle() {
+    if (_value != null) {
+      subject.add(_value = !_value!);
+      return this;
+    }
+    return null;
+  }
+}
+
+/// Foundation class used for custom `Types` outside the common native Dart
+/// types.
+/// For example, any custom "Model" class, like User().obs will use `Rx` as
+/// wrapper.
+class Rx<T> extends _RxImpl<T> {
+  Rx(T initial) : super(initial);
+
+  @override
+  dynamic toJson() {
+    try {
+      return (value as dynamic)?.toJson();
+    } on Exception catch (_) {
+      throw '$T has not method [toJson]';
+    }
+  }
+}
+
+class Rxn<T> extends Rx<T?> {
+  Rxn([T? initial]) : super(initial);
+
+  @override
+  dynamic toJson() {
+    try {
+      return (value as dynamic)?.toJson();
+    } on Exception catch (_) {
+      throw '$T has not method [toJson]';
+    }
+  }
+}
+
+extension StringExtension on String {
+  /// Returns a `RxString` with [this] `String` as initial value.
+  RxString get obs => RxString(this);
+}
+
+extension IntExtension on int {
+  /// Returns a `RxInt` with [this] `int` as initial value.
+  RxInt get obs => RxInt(this);
+}
+
+extension DoubleExtension on double {
+  /// Returns a `RxDouble` with [this] `double` as initial value.
+  RxDouble get obs => RxDouble(this);
+}
+
+extension BoolExtension on bool {
+  /// Returns a `RxBool` with [this] `bool` as initial value.
+  RxBool get obs => RxBool(this);
+}
+
+extension RxT<T> on T {
+  /// Returns a `Rx` instance with [this] `T` as initial value.
+  Rx<T> get obs => Rx<T>(this);
+}

packages/get/lib/get_rx/src/rx_types/rx_core/rx_interface.dart

@@ -0,0 +1,41 @@
+part of rx_types;
+
+/// This class is the foundation for all reactive (Rx) classes that makes Get
+/// so powerful.
+/// This interface is the contract that _RxImpl]<T> uses in all it's
+/// subclass.
+abstract class RxInterface<T> {
+  static RxInterface? proxy;
+
+  bool get canUpdate;
+
+  /// Adds a listener to stream
+  void addListener(GetStream<T> rxGetx);
+
+  /// Close the Rx Variable
+  void close();
+
+  /// Calls `callback` with current value, when the value changes.
+  StreamSubscription<T> listen(void Function(T event) onData,
+      {Function? onError, void Function()? onDone, bool? cancelOnError});
+
+  /// Avoids an unsafe usage of the `proxy`
+  static T notifyChildren<T>(RxNotifier observer, ValueGetter<T> builder) {
+    final _observer = RxInterface.proxy;
+    RxInterface.proxy = observer;
+    final result = builder();
+    if (!observer.canUpdate) {
+      RxInterface.proxy = _observer;
+      throw """
+      [Get] the improper use of a GetX has been detected. 
+      You should only use GetX or Obx for the specific widget that will be updated.
+      If you are seeing this error, you probably did not insert any observable variables into GetX/Obx 
+      or insert them outside the scope that GetX considers suitable for an update 
+      (example: GetX => HeavyWidget => variableObservable).
+      If you need to update a parent widget and a child widget, wrap each one in an Obx/GetX.
+      """;
+    }
+    RxInterface.proxy = _observer;
+    return result;
+  }
+}

packages/get/lib/get_rx/src/rx_types/rx_core/rx_string.dart

@@ -0,0 +1,289 @@
+part of rx_types;
+
+extension RxStringExt on Rx<String> {
+  String operator +(String val) => _value + val;
+
+  int compareTo(String other) {
+    return value.compareTo(other);
+  }
+
+  /// Returns true if this string ends with [other]. For example:
+  ///
+  ///     'Dart'.endsWith('t'); // true
+  bool endsWith(String other) {
+    return value.endsWith(other);
+  }
+
+  /// Returns true if this string starts with a match of [pattern].
+  bool startsWith(Pattern pattern, [int index = 0]) {
+    return value.startsWith(pattern, index);
+  }
+
+  /// Returns the position of the first match of [pattern] in this string
+  int indexOf(Pattern pattern, [int start = 0]) {
+    return value.indexOf(pattern, start);
+  }
+
+  /// Returns the starting position of the last match [pattern] in this string,
+  /// searching backward starting at [start], inclusive:
+  int lastIndexOf(Pattern pattern, [int? start]) {
+    return value.lastIndexOf(pattern, start);
+  }
+
+  /// Returns true if this string is empty.
+  bool get isEmpty => value.isEmpty;
+
+  /// Returns true if this string is not empty.
+  bool get isNotEmpty => !isEmpty;
+
+  /// Returns the substring of this string that extends from [startIndex],
+  /// inclusive, to [endIndex], exclusive
+  String substring(int startIndex, [int? endIndex]) {
+    return value.substring(startIndex, endIndex);
+  }
+
+  /// Returns the string without any leading and trailing whitespace.
+  String trim() {
+    return value.trim();
+  }
+
+  /// Returns the string without any leading whitespace.
+  ///
+  /// As [trim], but only removes leading whitespace.
+  String trimLeft() {
+    return value.trimLeft();
+  }
+
+  /// Returns the string without any trailing whitespace.
+  ///
+  /// As [trim], but only removes trailing whitespace.
+  String trimRight() {
+    return value.trimRight();
+  }
+
+  /// Pads this string on the left if it is shorter than [width].
+  ///
+  /// Return a new string that prepends [padding] onto this string
+  /// one time for each position the length is less than [width].
+  String padLeft(int width, [String padding = ' ']) {
+    return value.padLeft(width, padding);
+  }
+
+  /// Pads this string on the right if it is shorter than [width].
+
+  /// Return a new string that appends [padding] after this string
+  /// one time for each position the length is less than [width].
+  String padRight(int width, [String padding = ' ']) {
+    return value.padRight(width, padding);
+  }
+
+  /// Returns true if this string contains a match of [other]:
+  bool contains(Pattern other, [int startIndex = 0]) {
+    return value.contains(other, startIndex);
+  }
+
+  /// Replaces all substrings that match [from] with [replace].
+  String replaceAll(Pattern from, String replace) {
+    return value.replaceAll(from, replace);
+  }
+
+  /// Splits the string at matches of [pattern] and returns a list
+  /// of substrings.
+  List<String> split(Pattern pattern) {
+    return value.split(pattern);
+  }
+
+  /// Returns an unmodifiable list of the UTF-16 code units of this string.
+  List<int> get codeUnits => value.codeUnits;
+
+  /// Returns an [Iterable] of Unicode code-points of this string.
+  ///
+  /// If the string contains surrogate pairs, they are combined and returned
+  /// as one integer by this iterator. Unmatched surrogate halves are treated
+  /// like valid 16-bit code-units.
+  Runes get runes => value.runes;
+
+  /// Converts all characters in this string to lower case.
+  /// If the string is already in all lower case, this method returns `this`.
+  String toLowerCase() {
+    return value.toLowerCase();
+  }
+
+  /// Converts all characters in this string to upper case.
+  /// If the string is already in all upper case, this method returns `this`.
+  String toUpperCase() {
+    return value.toUpperCase();
+  }
+
+  Iterable<Match> allMatches(String string, [int start = 0]) {
+    return value.allMatches(string, start);
+  }
+
+  Match? matchAsPrefix(String string, [int start = 0]) {
+    return value.matchAsPrefix(string, start);
+  }
+}
+
+extension RxnStringExt on Rx<String?> {
+  String operator +(String val) => (_value ?? '') + val;
+
+  int? compareTo(String other) {
+    return value?.compareTo(other);
+  }
+
+  /// Returns true if this string ends with [other]. For example:
+  ///
+  ///     'Dart'.endsWith('t'); // true
+  bool? endsWith(String other) {
+    return value?.endsWith(other);
+  }
+
+  /// Returns true if this string starts with a match of [pattern].
+  bool? startsWith(Pattern pattern, [int index = 0]) {
+    return value?.startsWith(pattern, index);
+  }
+
+  /// Returns the position of the first match of [pattern] in this string
+  int? indexOf(Pattern pattern, [int start = 0]) {
+    return value?.indexOf(pattern, start);
+  }
+
+  /// Returns the starting position of the last match [pattern] in this string,
+  /// searching backward starting at [start], inclusive:
+  int? lastIndexOf(Pattern pattern, [int? start]) {
+    return value?.lastIndexOf(pattern, start);
+  }
+
+  /// Returns true if this string is empty.
+  bool? get isEmpty => value?.isEmpty;
+
+  /// Returns true if this string is not empty.
+  bool? get isNotEmpty => value?.isNotEmpty;
+
+  /// Returns the substring of this string that extends from [startIndex],
+  /// inclusive, to [endIndex], exclusive
+  String? substring(int startIndex, [int? endIndex]) {
+    return value?.substring(startIndex, endIndex);
+  }
+
+  /// Returns the string without any leading and trailing whitespace.
+  String? trim() {
+    return value?.trim();
+  }
+
+  /// Returns the string without any leading whitespace.
+  ///
+  /// As [trim], but only removes leading whitespace.
+  String? trimLeft() {
+    return value?.trimLeft();
+  }
+
+  /// Returns the string without any trailing whitespace.
+  ///
+  /// As [trim], but only removes trailing whitespace.
+  String? trimRight() {
+    return value?.trimRight();
+  }
+
+  /// Pads this string on the left if it is shorter than [width].
+  ///
+  /// Return a new string that prepends [padding] onto this string
+  /// one time for each position the length is less than [width].
+  String? padLeft(int width, [String padding = ' ']) {
+    return value?.padLeft(width, padding);
+  }
+
+  /// Pads this string on the right if it is shorter than [width].
+
+  /// Return a new string that appends [padding] after this string
+  /// one time for each position the length is less than [width].
+  String? padRight(int width, [String padding = ' ']) {
+    return value?.padRight(width, padding);
+  }
+
+  /// Returns true if this string contains a match of [other]:
+  bool? contains(Pattern other, [int startIndex = 0]) {
+    return value?.contains(other, startIndex);
+  }
+
+  /// Replaces all substrings that match [from] with [replace].
+  String? replaceAll(Pattern from, String replace) {
+    return value?.replaceAll(from, replace);
+  }
+
+  /// Splits the string at matches of [pattern] and returns a list
+  /// of substrings.
+  List<String>? split(Pattern pattern) {
+    return value?.split(pattern);
+  }
+
+  /// Returns an unmodifiable list of the UTF-16 code units of this string.
+  List<int>? get codeUnits => value?.codeUnits;
+
+  /// Returns an [Iterable] of Unicode code-points of this string.
+  ///
+  /// If the string contains surrogate pairs, they are combined and returned
+  /// as one integer by this iterator. Unmatched surrogate halves are treated
+  /// like valid 16-bit code-units.
+  Runes? get runes => value?.runes;
+
+  /// Converts all characters in this string to lower case.
+  /// If the string is already in all lower case, this method returns `this`.
+  String? toLowerCase() {
+    return value?.toLowerCase();
+  }
+
+  /// Converts all characters in this string to upper case.
+  /// If the string is already in all upper case, this method returns `this`.
+  String? toUpperCase() {
+    return value?.toUpperCase();
+  }
+
+  Iterable<Match>? allMatches(String string, [int start = 0]) {
+    return value?.allMatches(string, start);
+  }
+
+  Match? matchAsPrefix(String string, [int start = 0]) {
+    return value?.matchAsPrefix(string, start);
+  }
+}
+
+/// Rx class for `String` Type.
+class RxString extends Rx<String> implements Comparable<String>, Pattern {
+  RxString(String initial) : super(initial);
+
+  @override
+  Iterable<Match> allMatches(String string, [int start = 0]) {
+    return value.allMatches(string, start);
+  }
+
+  @override
+  Match? matchAsPrefix(String string, [int start = 0]) {
+    return value.matchAsPrefix(string, start);
+  }
+
+  @override
+  int compareTo(String other) {
+    return value.compareTo(other);
+  }
+}
+
+/// Rx class for `String` Type.
+class RxnString extends Rx<String?> implements Comparable<String>, Pattern {
+  RxnString([String? initial]) : super(initial);
+
+  @override
+  Iterable<Match> allMatches(String string, [int start = 0]) {
+    return value!.allMatches(string, start);
+  }
+
+  @override
+  Match? matchAsPrefix(String string, [int start = 0]) {
+    return value!.matchAsPrefix(string, start);
+  }
+
+  @override
+  int compareTo(String other) {
+    return value!.compareTo(other);
+  }
+}

packages/get/lib/get_rx/src/rx_types/rx_iterables/rx_list.dart

@@ -0,0 +1,172 @@
+part of rx_types;
+
+/// Create a list similar to `List<T>`
+class RxList<E> extends ListMixin<E>
+    with NotifyManager<List<E>>, RxObjectMixin<List<E>>
+    implements RxInterface<List<E>> {
+  RxList([List<E> initial = const []]) {
+    _value = List.from(initial);
+  }
+
+  factory RxList.filled(int length, E fill, {bool growable = false}) {
+    return RxList(List.filled(length, fill, growable: growable));
+  }
+
+  factory RxList.empty({bool growable = false}) {
+    return RxList(List.empty(growable: growable));
+  }
+
+  /// Creates a list containing all [elements].
+  factory RxList.from(Iterable elements, {bool growable = true}) {
+    return RxList(List.from(elements, growable: growable));
+  }
+
+  /// Creates a list from [elements].
+  factory RxList.of(Iterable<E> elements, {bool growable = true}) {
+    return RxList(List.of(elements, growable: growable));
+  }
+
+  /// Generates a list of values.
+  factory RxList.generate(int length, E generator(int index),
+      {bool growable = true}) {
+    return RxList(List.generate(length, generator, growable: growable));
+  }
+
+  /// Creates an unmodifiable list containing all [elements].
+  factory RxList.unmodifiable(Iterable elements) {
+    return RxList(List.unmodifiable(elements));
+  }
+
+  @override
+  Iterator<E> get iterator => value.iterator;
+
+  @override
+  void operator []=(int index, E val) {
+    _value[index] = val;
+    refresh();
+  }
+
+  /// Special override to push() element(s) in a reactive way
+  /// inside the List,
+  @override
+  RxList<E> operator +(Iterable<E> val) {
+    addAll(val);
+    refresh();
+    return this;
+  }
+
+  @override
+  E operator [](int index) {
+    return value[index];
+  }
+
+  @override
+  void add(E item) {
+    _value.add(item);
+    refresh();
+  }
+
+  @override
+  void addAll(Iterable<E> item) {
+    _value.addAll(item);
+    refresh();
+  }
+
+  @override
+  void removeWhere(bool test(E element)) {
+    _value.removeWhere(test);
+    refresh();
+  }
+
+  @override
+  void retainWhere(bool test(E element)) {
+    _value.retainWhere(test);
+    refresh();
+  }
+
+  @override
+  int get length => value.length;
+
+  @override
+  @protected
+  List<E> get value {
+    RxInterface.proxy?.addListener(subject);
+    return _value;
+  }
+
+  @override
+  set length(int newLength) {
+    _value.length = newLength;
+    refresh();
+  }
+
+  @override
+  void insertAll(int index, Iterable<E> iterable) {
+    _value.insertAll(index, iterable);
+    refresh();
+  }
+
+  @override
+  Iterable<E> get reversed => value.reversed;
+
+  @override
+  Iterable<E> where(bool Function(E) test) {
+    return value.where(test);
+  }
+
+  @override
+  Iterable<T> whereType<T>() {
+    return value.whereType<T>();
+  }
+
+  @override
+  void sort([int compare(E a, E b)?]) {
+    _value.sort(compare);
+    refresh();
+  }
+}
+
+extension ListExtension<E> on List<E> {
+  RxList<E> get obs => RxList<E>(this);
+
+  /// Add [item] to [List<E>] only if [item] is not null.
+  void addNonNull(E item) {
+    if (item != null) add(item);
+  }
+
+  // /// Add [Iterable<E>] to [List<E>] only if [Iterable<E>] is not null.
+  // void addAllNonNull(Iterable<E> item) {
+  //   if (item != null) addAll(item);
+  // }
+
+  /// Add [item] to List<E> only if [condition] is true.
+  void addIf(dynamic condition, E item) {
+    if (condition is Condition) condition = condition();
+    if (condition is bool && condition) add(item);
+  }
+
+  /// Adds [Iterable<E>] to [List<E>] only if [condition] is true.
+  void addAllIf(dynamic condition, Iterable<E> items) {
+    if (condition is Condition) condition = condition();
+    if (condition is bool && condition) addAll(items);
+  }
+
+  /// Replaces all existing items of this list with [item]
+  void assign(E item) {
+    // if (this is RxList) {
+    //   (this as RxList)._value;
+    // }
+
+    clear();
+    add(item);
+  }
+
+  /// Replaces all existing items of this list with [items]
+  void assignAll(Iterable<E> items) {
+    // if (this is RxList) {
+    //   (this as RxList)._value;
+    // }
+    clear();
+    addAll(items);
+  }
+}

packages/get/lib/get_rx/src/rx_types/rx_iterables/rx_map.dart

@@ -0,0 +1,108 @@
+part of rx_types;
+
+class RxMap<K, V> extends MapMixin<K, V>
+    with NotifyManager<Map<K, V>>, RxObjectMixin<Map<K, V>>
+    implements RxInterface<Map<K, V>> {
+  RxMap([Map<K, V> initial = const {}]) {
+    _value = Map.from(initial);
+  }
+
+  factory RxMap.from(Map<K, V> other) {
+    return RxMap(Map.from(other));
+  }
+
+  /// Creates a [LinkedHashMap] with the same keys and values as [other].
+  factory RxMap.of(Map<K, V> other) {
+    return RxMap(Map.of(other));
+  }
+
+  ///Creates an unmodifiable hash based map containing the entries of [other].
+  factory RxMap.unmodifiable(Map<dynamic, dynamic> other) {
+    return RxMap(Map.unmodifiable(other));
+  }
+
+  /// Creates an identity map with the default implementation, [LinkedHashMap].
+  factory RxMap.identity() {
+    return RxMap(Map.identity());
+  }
+
+  @override
+  V? operator [](Object? key) {
+    return value[key as K];
+  }
+
+  @override
+  void operator []=(K key, V value) {
+    _value[key] = value;
+    refresh();
+  }
+
+  @override
+  void clear() {
+    _value.clear();
+    refresh();
+  }
+
+  @override
+  Iterable<K> get keys => value.keys;
+
+  @override
+  V? remove(Object? key) {
+    final val = _value.remove(key);
+    refresh();
+    return val;
+  }
+
+  @override
+  @protected
+  Map<K, V> get value {
+    RxInterface.proxy?.addListener(subject);
+    return _value;
+  }
+}
+
+extension MapExtension<K, V> on Map<K, V> {
+  RxMap<K, V> get obs {
+    return RxMap<K, V>(this);
+  }
+
+  void addIf(dynamic condition, K key, V value) {
+    if (condition is Condition) condition = condition();
+    if (condition is bool && condition) {
+      this[key] = value;
+    }
+  }
+
+  void addAllIf(dynamic condition, Map<K, V> values) {
+    if (condition is Condition) condition = condition();
+    if (condition is bool && condition) addAll(values);
+  }
+
+  void assign(K key, V val) {
+    if (this is RxMap) {
+      final map = (this as RxMap);
+      // map._value;
+      map._value.clear();
+      this[key] = val;
+    } else {
+      clear();
+      this[key] = val;
+    }
+  }
+
+  void assignAll(Map<K, V> val) {
+    if (val is RxMap && this is RxMap) {
+      if ((val as RxMap)._value == (this as RxMap)._value) return;
+    }
+    if (this is RxMap) {
+      final map = (this as RxMap);
+      if (map._value == val) return;
+      map._value = val;
+      map.refresh();
+    } else {
+      if (this == val) return;
+      clear();
+      addAll(val);
+    }
+  }
+}

packages/get/lib/get_rx/src/rx_types/rx_iterables/rx_set.dart

@@ -0,0 +1,151 @@
+part of rx_types;
+
+class RxSet<E> extends SetMixin<E>
+    with NotifyManager<Set<E>>, RxObjectMixin<Set<E>>
+    implements RxInterface<Set<E>> {
+  RxSet([Set<E> initial = const {}]) {
+    _value = Set.from(initial);
+  }
+
+  /// Special override to push() element(s) in a reactive way
+  /// inside the List,
+  RxSet<E> operator +(Set<E> val) {
+    addAll(val);
+    refresh();
+    return this;
+  }
+
+  void update(void fn(Iterable<E>? value)) {
+    fn(value);
+    refresh();
+  }
+
+  @override
+  @protected
+  Set<E> get value {
+    RxInterface.proxy?.addListener(subject);
+    return _value;
+  }
+
+  @override
+  @protected
+  set value(Set<E> val) {
+    if (_value == val) return;
+    _value = val;
+    refresh();
+  }
+
+  @override
+  bool add(E value) {
+    final val = _value.add(value);
+    refresh();
+    return val;
+  }
+
+  @override
+  bool contains(Object? element) {
+    return value.contains(element);
+  }
+
+  @override
+  Iterator<E> get iterator => value.iterator;
+
+  @override
+  int get length => value.length;
+
+  @override
+  E? lookup(Object? object) {
+    return value.lookup(object);
+  }
+
+  @override
+  bool remove(Object? item) {
+    var hasRemoved = _value.remove(item);
+    if (hasRemoved) {
+      refresh();
+    }
+    return hasRemoved;
+  }
+
+  @override
+  Set<E> toSet() {
+    return value.toSet();
+  }
+
+  @override
+  void addAll(Iterable<E> item) {
+    _value.addAll(item);
+    refresh();
+  }
+
+  @override
+  void clear() {
+    _value.clear();
+    refresh();
+  }
+
+  @override
+  void removeAll(Iterable<Object?> elements) {
+    _value.removeAll(elements);
+    refresh();
+  }
+
+  @override
+  void retainAll(Iterable<Object?> elements) {
+    _value.retainAll(elements);
+    refresh();
+  }
+
+  @override
+  void retainWhere(bool Function(E) E) {
+    _value.retainWhere(E);
+    refresh();
+  }
+}
+
+extension SetExtension<E> on Set<E> {
+  RxSet<E> get obs {
+    return RxSet<E>(<E>{})..addAll(this);
+  }
+
+  // /// Add [item] to [List<E>] only if [item] is not null.
+  // void addNonNull(E item) {
+  //   if (item != null) add(item);
+  // }
+
+  // /// Add [Iterable<E>] to [List<E>] only if [Iterable<E>] is not null.
+  // void addAllNonNull(Iterable<E> item) {
+  //   if (item != null) addAll(item);
+  // }
+
+  /// Add [item] to [List<E>] only if [condition] is true.
+  void addIf(dynamic condition, E item) {
+    if (condition is Condition) condition = condition();
+    if (condition is bool && condition) add(item);
+  }
+
+  /// Adds [Iterable<E>] to [List<E>] only if [condition] is true.
+  void addAllIf(dynamic condition, Iterable<E> items) {
+    if (condition is Condition) condition = condition();
+    if (condition is bool && condition) addAll(items);
+  }
+
+  /// Replaces all existing items of this list with [item]
+  void assign(E item) {
+    // if (this is RxSet) {
+    //   (this as RxSet)._value;
+    // }
+
+    clear();
+    add(item);
+  }
+
+  /// Replaces all existing items of this list with [items]
+  void assignAll(Iterable<E> items) {
+    // if (this is RxSet) {
+    //   (this as RxSet)._value;
+    // }
+    clear();
+    addAll(items);
+  }
+}

packages/get/lib/get_rx/src/rx_types/rx_types.dart

@@ -0,0 +1,17 @@
+library rx_types;
+
+import 'dart:async';
+import 'dart:collection';
+
+import 'package:flutter/foundation.dart';
+import '../rx_stream/rx_stream.dart';
+import '../rx_typedefs/rx_typedefs.dart';
+
+part 'rx_core/rx_impl.dart';
+part 'rx_core/rx_interface.dart';
+part 'rx_core/rx_num.dart';
+part 'rx_core/rx_string.dart';
+
+part 'rx_iterables/rx_list.dart';
+part 'rx_iterables/rx_set.dart';
+part 'rx_iterables/rx_map.dart';

packages/get/lib/get_rx/src/rx_workers/rx_workers.dart

@@ -0,0 +1,273 @@
+import 'dart:async';
+
+import '../../../get_core/get_core.dart';
+import '../rx_types/rx_types.dart';
+import 'utils/debouncer.dart';
+
+bool _conditional(dynamic condition) {
+  if (condition == null) return true;
+  if (condition is bool) return condition;
+  if (condition is bool Function()) return condition();
+  return true;
+}
+
+typedef WorkerCallback<T> = Function(T callback);
+
+class Workers {
+  Workers(this.workers);
+  final List<Worker> workers;
+
+  void dispose() {
+    for (final worker in workers) {
+      if (!worker._disposed) {
+        worker.dispose();
+      }
+    }
+  }
+}
+
+///
+/// Called every time [listener] changes. As long as the [condition]
+/// returns true.
+///
+/// Sample:
+/// Every time increment() is called, ever() will process the [condition]
+/// (can be a [bool] expression or a `bool Function()`), and only call
+/// the callback when [condition] is true.
+/// In our case, only when count is bigger to 5. In order to "dispose"
+/// this Worker
+/// that will run forever, we made a `worker` variable. So, when the count value
+/// reaches 10, the worker gets disposed, and releases any memory resources.
+///
+/// ```
+/// // imagine some counter widget...
+///
+/// class _CountController extends GetxController {
+///   final count = 0.obs;
+///   Worker worker;
+///
+///   void onInit() {
+///     worker = ever(count, (value) {
+///       print('counter changed to: $value');
+///       if (value == 10) worker.dispose();
+///     }, condition: () => count > 5);
+///   }
+///
+///   void increment() => count + 1;
+/// }
+/// ```
+Worker ever<T>(
+  RxInterface<T> listener,
+  WorkerCallback<T> callback, {
+  dynamic condition = true,
+  Function? onError,
+  void Function()? onDone,
+  bool? cancelOnError,
+}) {
+  StreamSubscription sub = listener.listen(
+    (event) {
+      if (_conditional(condition)) callback(event);
+    },
+    onError: onError,
+    onDone: onDone,
+    cancelOnError: cancelOnError,
+  );
+  return Worker(sub.cancel, '[ever]');
+}
+
+/// Similar to [ever], but takes a list of [listeners], the condition
+/// for the [callback] is common to all [listeners],
+/// and the [callback] is executed to each one of them. The [Worker] is
+/// common to all, so `worker.dispose()` will cancel all streams.
+Worker everAll(
+  List<RxInterface> listeners,
+  WorkerCallback callback, {
+  dynamic condition = true,
+  Function? onError,
+  void Function()? onDone,
+  bool? cancelOnError,
+}) {
+  final evers = <StreamSubscription>[];
+  for (var i in listeners) {
+    final sub = i.listen(
+      (event) {
+        if (_conditional(condition)) callback(event);
+      },
+      onError: onError,
+      onDone: onDone,
+      cancelOnError: cancelOnError,
+    );
+    evers.add(sub);
+  }
+
+  Future<void> cancel() {
+    for (var i in evers) {
+      i.cancel();
+    }
+    return Future.value(() {});
+  }
+
+  return Worker(cancel, '[everAll]');
+}
+
+/// `once()` will execute only 1 time when [condition] is met and cancel
+/// the subscription to the [listener] stream right after that.
+/// [condition] defines when [callback] is called, and
+/// can be a [bool] or a `bool Function()`.
+///
+/// Sample:
+/// ```
+///  class _CountController extends GetxController {
+///   final count = 0.obs;
+///   Worker worker;
+///
+///   @override
+///   Future<void> onInit() async {
+///     worker = once(count, (value) {
+///       print("counter reached $value before 3 seconds.");
+///     }, condition: () => count() > 2);
+///     3.delay(worker.dispose);
+///   }
+///   void increment() => count + 1;
+/// }
+///```
+Worker once<T>(
+  RxInterface<T> listener,
+  WorkerCallback<T> callback, {
+  dynamic condition = true,
+  Function? onError,
+  void Function()? onDone,
+  bool? cancelOnError,
+}) {
+  late Worker ref;
+  StreamSubscription? sub;
+  sub = listener.listen(
+    (event) {
+      if (!_conditional(condition)) return;
+      ref._disposed = true;
+      ref._log('called');
+      sub?.cancel();
+      callback(event);
+    },
+    onError: onError,
+    onDone: onDone,
+    cancelOnError: cancelOnError,
+  );
+  ref = Worker(sub.cancel, '[once]');
+  return ref;
+}
+
+/// Ignore all changes in [listener] during [time] (1 sec by default) or until
+/// [condition] is met (can be a [bool] expression or a `bool Function()`),
+/// It brings the 1st "value" since the period of time, so
+/// if you click a counter button 3 times in 1 sec, it will show you "1"
+/// (after 1 sec of the first press)
+/// click counter 3 times in 1 sec, it will show you "4" (after 1 sec)
+/// click counter 2 times in 1 sec, it will show you "7" (after 1 sec).
+///
+/// Sample:
+/// // wait 1 sec each time an event starts, only if counter is lower than 20.
+/// worker = interval(
+///    count,
+///    (value) => print(value),
+///    time: 1.seconds,
+///    condition: () => count < 20,
+/// );
+/// ```
+Worker interval<T>(
+  RxInterface<T> listener,
+  WorkerCallback<T> callback, {
+  Duration time = const Duration(seconds: 1),
+  dynamic condition = true,
+  Function? onError,
+  void Function()? onDone,
+  bool? cancelOnError,
+}) {
+  var debounceActive = false;
+  StreamSubscription sub = listener.listen(
+    (event) async {
+      if (debounceActive || !_conditional(condition)) return;
+      debounceActive = true;
+      await Future.delayed(time);
+      debounceActive = false;
+      callback(event);
+    },
+    onError: onError,
+    onDone: onDone,
+    cancelOnError: cancelOnError,
+  );
+  return Worker(sub.cancel, '[interval]');
+}
+
+/// [debounce] is similar to [interval], but sends the last value.
+/// Useful for Anti DDos, every time the user stops typing for 1 second,
+/// for instance.
+/// When [listener] emits the last "value", when [time] hits,
+/// it calls [callback] with the last "value" emitted.
+///
+/// Sample:
+///
+/// ```
+/// worker = debounce(
+///      count,
+///      (value) {
+///        print(value);
+///        if( value > 20 ) worker.dispose();
+///      },
+///      time: 1.seconds,
+///    );
+///  }
+///  ```
+Worker debounce<T>(
+  RxInterface<T> listener,
+  WorkerCallback<T> callback, {
+  Duration? time,
+  Function? onError,
+  void Function()? onDone,
+  bool? cancelOnError,
+}) {
+  final _debouncer =
+      Debouncer(delay: time ?? const Duration(milliseconds: 800));
+  StreamSubscription sub = listener.listen(
+    (event) {
+      _debouncer(() {
+        callback(event);
+      });
+    },
+    onError: onError,
+    onDone: onDone,
+    cancelOnError: cancelOnError,
+  );
+  return Worker(sub.cancel, '[debounce]');
+}
+
+class Worker {
+  Worker(this.worker, this.type);
+
+  /// subscription.cancel() callback
+  final Future<void> Function() worker;
+
+  /// type of worker (debounce, interval, ever)..
+  final String type;
+  bool _disposed = false;
+
+  bool get disposed => _disposed;
+
+  //final bool _verbose = true;
+  void _log(String msg) {
+    //  if (!_verbose) return;
+    Get.log('$runtimeType $type $msg');
+  }
+
+  void dispose() {
+    if (_disposed) {
+      _log('already disposed');
+      return;
+    }
+    _disposed = true;
+    worker();
+    _log('disposed');
+  }
+
+  void call() => dispose();
+}

packages/get/lib/get_rx/src/rx_workers/utils/debouncer.dart

@@ -0,0 +1,27 @@
+import 'dart:async';
+
+/// This "function" class is the implementation of `debouncer()` Worker.
+/// It calls the function passed after specified [delay] parameter.
+/// Example:
+/// ```
+/// final delayed = Debouncer( delay: Duration( seconds: 1 )) ;
+/// print( 'the next function will be called after 1 sec' );
+/// delayed( () => print( 'called after 1 sec' ));
+/// ```
+class Debouncer {
+  final Duration? delay;
+  Timer? _timer;
+
+  Debouncer({this.delay});
+
+  void call(void Function() action) {
+    _timer?.cancel();
+    _timer = Timer(delay!, action);
+  }
+
+  /// Notifies if the delayed call is active.
+  bool get isRunning => _timer?.isActive ?? false;
+
+  /// Cancel the current delayed call.
+  void cancel() => _timer?.cancel();
+}