Drop reference to cached items asap in to_xxx (#509)

This reduces memory consumption on cases where the cached items are
quite big and: The observable is not disposed immediately or the
completion callback triggers other computations.

rx/core/operators/todict.py

@@ -39,7 +39,9 @@ def on_next(x: Any) -> None:
                 m[key] = element
 
             def on_completed() -> None:
+                nonlocal m
                 observer.on_next(m)
+                m = dict()
                 observer.on_completed()
 
             return source.subscribe_(on_next, observer.on_error, on_completed, scheduler)

rx/core/operators/tofuture.py

@@ -41,10 +41,12 @@ def on_error(err):
             future.set_exception(err)
 
         def on_completed():
+            nonlocal last_value
             if has_value:
                 future.set_result(last_value)
             else:
                 future.set_exception(SequenceContainsNoElementsError())
+            last_value = None
 
         source.subscribe_(on_next, on_error, on_completed, scheduler=scheduler)
 

rx/core/operators/toiterable.py

@@ -20,7 +20,9 @@ def on_next(item):
                 queue.append(item)
 
             def on_completed():
+                nonlocal queue
                 observer.on_next(queue)
+                queue = []
                 observer.on_completed()
 
             return source.subscribe_(on_next, observer.on_error, on_completed, scheduler)

rx/core/operators/toset.py

@@ -14,7 +14,9 @@ def subscribe(observer, scheduler=None):
             s = set()
 
             def on_completed():
+                nonlocal s
                 observer.on_next(s)
+                s = set()
                 observer.on_completed()
 
             return source.subscribe_(s.add, observer.on_error, on_completed, scheduler)