Closed ZhZhZhen closed 2 years ago
You can use CompositeSubscription
to manage subscriptions easily
In addition, you can use my package disposebag and flutter_disposebag
You can use
CompositeSubscription
to manage subscriptions easily
thank you, it is useful to manage subscription. but i want to remove finished/cancelled request automatically to avoid CompositeSubscription._subscriptionsList too large, may still need to create a StreamController
Here is my code that removes subscription on done or on cancel
import 'dart:async';
import 'package:rxdart/rxdart.dart';
void main() async {
final t = TestViewModel();
t.beginCounter();
await Future.delayed(Duration(seconds: 11));
print('END');
}
class Repository {
Stream<int> counter() async* {
for (int i = 0; i < 10; i++) {
await Future.delayed(Duration(seconds: 1));
yield i;
}
}
}
class TestViewModel with SubscriptionBind {
var repo = Repository();
void beginCounter() {
repo
.counter()
.bindLife(this) //i want to cancel Stream from here
.map((event) => event) //emulate other operators
.listen((event) {
print('event:$event');
});
}
}
mixin SubscriptionBind {
final _compositeSubscription = CompositeSubscription();
void dispose() {
_compositeSubscription.dispose();
print('disposed in $this, length=${_compositeSubscription.length}');
}
Stream<T> obsStream<T>(Stream<T> stream) {
final controller = StreamController<T>(sync: true);
late StreamSubscription<T> subscription;
controller.onListen = () {
subscription = stream.listen(
controller.add,
onError: controller.addError,
onDone: () {
_compositeSubscription.remove(
subscription,
shouldCancel: false,
);
print('onDone: length=${_compositeSubscription.length}');
controller.close();
},
)..addTo(_compositeSubscription);
print('onListen: length=${_compositeSubscription.length}');
};
controller.onCancel = () {
final future = _compositeSubscription.remove(
subscription,
shouldCancel: true,
);
print('onCancel: length=${_compositeSubscription.length}');
return future;
};
if (!stream.isBroadcast) {
controller.onPause = () => subscription.pause();
controller.onResume = () => subscription.resume();
}
return controller.stream;
}
}
extension StreamLifecycleExt<T> on Stream<T> {
Stream<T> bindLife(SubscriptionBind vm) {
return vm.obsStream(this);
}
}
@hoc081098 thank you, considering a lot of things that I didn't think about, benefit me a lot.
i use ChangeNotifierProvider+ChangeNotifier to manage state in Flutter,i want to cancel all request when the ChangeNotifier.dispose() executed. And i wish to remove the Subscription when request is finished/cancelled. I'm not sure if this is a good approach, is there any potential risk with this code
here is my code.