This feels a bit hacky at the moment. TODO: clean this up.
(self, loop, new_program)
| 453 | return self._variables.get(key, default) |
| 454 | |
| 455 | def _stream_run(self, loop, new_program): |
| 456 | """This feels a bit hacky at the moment. TODO: clean this up.""" |
| 457 | |
| 458 | # add the program execution to the event loop |
| 459 | execute_task = loop.create_task(new_program.execute()) |
| 460 | new_program._tasks.append(execute_task) |
| 461 | |
| 462 | # run the event loop until the program is done executing |
| 463 | while new_program._executor is not None: |
| 464 | try: |
| 465 | loop.run_until_complete( |
| 466 | execute_task |
| 467 | ) # this will stop each time the program wants to emit a new state |
| 468 | except RuntimeError as e: |
| 469 | # we don't mind that the task is not yet done, we will restart the loop |
| 470 | if str(e) != "Event loop stopped before Future completed.": |
| 471 | raise e |
| 472 | if getattr(loop, "_stopping", False): |
| 473 | loop._stopping = False # clean up the stopping flag |
| 474 | if new_program._executor is not None and new_program._executor.executing: |
| 475 | try: |
| 476 | yield new_program |
| 477 | except GeneratorExit: |
| 478 | # this will cause the program to stop executing and finish as a valid partial execution |
| 479 | if new_program._executor.executing: |
| 480 | new_program._executor.executing = False |
| 481 | yield new_program |
| 482 | |
| 483 | # cancel all tasks and close the loop |
| 484 | for task in self._tasks: |
| 485 | task.cancel() |
| 486 | loop.run_until_complete( |
| 487 | asyncio.sleep(0) |
| 488 | ) # give the loop a chance to cancel the tasks |
| 489 | |
| 490 | # TODO: do we really want to close the loop? what if it is used by others? |
| 491 | loop.close() # we are done with the loop (note that the loop is already stopped) |
| 492 | |
| 493 | async def _stream_run_async(self): |
| 494 | # run the event loop until the program is done executing |