fixing dispose for ToFuture

semiversus · Jun 15, 2018 · 2af3278 · 2af3278
1 parent 309a73a
commit 2af3278
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/broqer/op/to_future.py b/broqer/op/to_future.py
@@ -50,12 +50,16 @@ def _timeout(self):
         self.set_exception(asyncio.TimeoutError)
 
     def _future_done(self, future):
+        if self._disposable is not None:
+            self._disposable.dispose()
+            self._disposable = None
         if self._timeout_handle is not None:
             self._timeout_handle.cancel()
 
     def emit(self, *args: Any, who: Publisher) -> None:
         if self._disposable is not None:
             self._disposable.dispose()
+            self._disposable = None
         if len(args) == 1:
             self.set_result(args[0])
         else: