Step 8 - Extend the Loop

As a next step we want to extend the loop so that we can get the current loop.

Loop v2

from typing import Any, Generator


class Loop:
    """Loop v2"""

    _instance: "Loop" = None

    def __init__(self):
        self._running = False

    @classmethod
    def get_current_loop(cls) -> "Loop":
        if not cls._instance:
            cls._instance = Loop()
        return cls._instance

    def run(self, coroutine: Generator[Any, None, Any]) -> Any:
        """Run a coroutine"""
        self._running = True
        step = 1
        while self._running:
            print("Loop step", step)
            try:
                next(coroutine)
                step += 1
            except StopIteration as e:
                self._running = False
                return e.value

    def stop(self) -> None:
        """Stop running the loop"""
        self._running = False

The loop has been converted into a class that uses a singleton pattern to return the current loop. The current loop is now accessible via Loop.get_current_loop(). A coroutine can be started by calling the run method on the loop instance. The state of the loop (running/not running) is tracked in the private _running property.

from future import Future

from loop import Loop


def some_result(value):
    future = Future()
    future.set_result(value)
    return (yield from future)


def add(coroutine1, coroutine2):
    x = yield from coroutine1
    y = yield from coroutine2
    return x + y


def main():
    return (yield from add(some_result(1), some_result(2)))


loop = Loop.get_current_loop()
result = loop.run(main())
print("Loop finished with result", result)

Output:

Loop step 1
Loop step 2
Loop step 3
Loop finished with result 3

Summary

• The loop will not be instantiated by the application code directly.

• Instead the current loop is requested at a single API point.

• The loop could decide whether it returns a new instance or not.