fix handle_cancelled_feeder() and handle_cancelled_feeder()
handle_cancelled_feeder() should not modify done observer
handle_cancelled_feeder() may be called from running event loop
@@ -149,10 +149,9 @@ def is_feeder(task):
def handle_cancelled_feeder(connection_observer, observer_lock, logger, future):
- if future.cancelled() and not connection_observer.cancelled():
+ if future.cancelled() and not connection_observer.done():
logger.debug("cancelled {}".format(future))
with observer_lock:
- if not connection_observer.done():
logger.debug("cancelling {}".format(connection_observer))
connection_observer.cancel()
@@ -167,6 +166,9 @@ def cancel_remaining_feeders(loop, logger_name="moler.runner.asyncio"):
for feeder in remaining:
logger.debug(" remaining {}:{}".format(instance_id(feeder), feeder))
remaining_tasks.cancel()
+ if loop.is_running():
+ remaining_tasks.add_done_callback(lambda t: loop.stop())
+ else:
# Keep the event loop running until it is either destroyed or all tasks have really terminated
loop.run_until_complete(remaining_tasks)
