fix Rx variables out of GetXController memory leak

@@ -43,7 +43,5 @@ class MockClient extends HttpRequestBase {
   }
 
   @override
-  void close() {
-    // TODO: implement close
-  }
+  void close() {}
 }

@@ -31,7 +31,7 @@ extension ExtensionSnackbar on GetInterface {
     Color leftBarIndicatorColor,
     List<BoxShadow> boxShadows,
     Gradient backgroundGradient,
-    FlatButton mainButton,
+    Widget mainButton,
     OnTap onTap,
     Duration duration = const Duration(seconds: 3),
     bool isDismissible = true,

@@ -91,8 +91,8 @@ class GetBar<T extends Object> extends StatefulWidget {
   /// An option to animate the icon (if present). Defaults to true.
   final bool shouldIconPulse;
 
-  /// A [FlatButton] widget if you need an action from the user.
-  final FlatButton mainButton;
+  /// A [TextButton] widget if you need an action from the user.
+  final Widget mainButton;
 
   /// A callback that registers the user's click anywhere.
   /// An alternative to [mainButton]
@@ -659,7 +659,7 @@ Set either a message or messageText""");
     );
   }
 
-  FlatButton _getMainActionButton() {
+  Widget _getMainActionButton() {
     return widget.mainButton;
   }
 }

@@ -116,7 +116,7 @@ mixin RxObjectMixin<T> on NotifyManager<T> {
   /// Closing the subscription will happen automatically when the observer
   /// Widget ([GetX] or [Obx]) gets unmounted from the Widget tree.
   void bindStream(Stream<T> stream) {
-    _subscriptions.add(stream.listen((va) => value = va));
+    _subscriptions[subject].add(stream.listen((va) => value = va));
   }
 }
 
@@ -124,21 +124,19 @@ class RxNotifier<T> = RxInterface<T> with NotifyManager<T>;
 
 mixin NotifyManager<T> {
   GetStream<T> subject = GetStream<T>();
-  final _subscriptions = <StreamSubscription>[];
+  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.contains(rxGetx.listen)) {
-      return;
+    if (!_subscriptions.containsKey(rxGetx)) {
+      final subs = rxGetx.listen(subject.add);
+      final listSubscriptions =
+          _subscriptions[rxGetx] ??= <StreamSubscription>[];
+      listSubscriptions.add(subs);
     }
-
-    final subs = rxGetx.listen((data) {
-      subject.add(data);
-    });
-    _subscriptions.add(subs);
   }
 
   StreamSubscription<T> listen(
@@ -152,9 +150,12 @@ mixin NotifyManager<T> {
 
   /// 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();
   }