fix concurrent modification

@@ -55,6 +55,8 @@ You cannot ${isClosed ? "close" : "add events to"} a closed stream.''';
     listenable.addSubscription(subs);
     return subs;
   }
+
+  Stream<T> get stream => GetStreamTransformation(listenable);
 }
 
 class LightListenable<T> {
@@ -125,7 +127,10 @@ class LightSubscription<T> extends StreamSubscription<T> {
   bool cancelOnError = false;
 
   @override
-  Future<void> cancel() async => listener.removeSubscription(this);
+  Future<void> cancel() {
+    listener.removeSubscription(this);
+    return Future.value();
+  }
 
   OnData<T> _data;
 

 library rx_stream;
 
 import 'dart:async';
+import 'package:flutter/scheduler.dart';
+
 import '../rx_typedefs/rx_typedefs.dart';
 
 part 'get_stream.dart';

@@ -11,6 +11,8 @@ bool _conditional(dynamic condition) {
   return true;
 }
 
+typedef WorkerCallback<T> = Function(T callback);
+
 ///
 /// Called every time [listener] changes. As long as the [condition]
 /// returns true.
@@ -41,7 +43,7 @@ bool _conditional(dynamic condition) {
 ///   void increment() => count + 1;
 /// }
 /// ```
-Worker ever<T>(RxInterface<T> listener, Function(T) callback,
+Worker ever<T>(RxInterface<T> listener, WorkerCallback<T> callback,
     {dynamic condition = true}) {
   StreamSubscription sub = listener.subject.listen((event) {
     if (_conditional(condition)) callback(event);
@@ -53,7 +55,7 @@ Worker ever<T>(RxInterface<T> listener, Function(T) callback,
 /// 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, Function(dynamic) callback,
+Worker everAll(List<RxInterface> listeners, WorkerCallback callback,
     {dynamic condition = true}) {
   final evers = <StreamSubscription>[];
   for (var i in listeners) {
@@ -94,7 +96,7 @@ Worker everAll(List<RxInterface> listeners, Function(dynamic) callback,
 ///   void increment() => count + 1;
 /// }
 ///```
-Worker once<T>(RxInterface<T> listener, Function(T) callback,
+Worker once<T>(RxInterface<T> listener, WorkerCallback<T> callback,
     {dynamic condition}) {
   Worker ref;
   StreamSubscription sub;
@@ -102,8 +104,10 @@ Worker once<T>(RxInterface<T> listener, Function(T) callback,
     if (!_conditional(condition)) return;
     ref._disposed = true;
     ref._log('called');
-    sub?.cancel();
     callback(event);
+    Timer.run(() {
+      sub?.cancel();
+    });
   });
   ref = Worker(sub.cancel, '[once]');
   return ref;
@@ -126,7 +130,7 @@ Worker once<T>(RxInterface<T> listener, Function(T) callback,
 ///    condition: () => count < 20,
 /// );
 /// ```
-Worker interval<T>(RxInterface<T> listener, Function(T) callback,
+Worker interval<T>(RxInterface<T> listener, WorkerCallback<T> callback,
     {Duration time = const Duration(seconds: 1), dynamic condition = true}) {
   var debounceActive = false;
   time ??= const Duration(seconds: 1);
@@ -159,7 +163,7 @@ Worker interval<T>(RxInterface<T> listener, Function(T) callback,
 ///    );
 ///  }
 ///  ```
-Worker debounce<T>(RxInterface<T> listener, Function(T) callback,
+Worker debounce<T>(RxInterface<T> listener, WorkerCallback<T> callback,
     {Duration time}) {
   final _debouncer =
       Debouncer(delay: time ?? const Duration(milliseconds: 800));