Helper to run a background tulip loop in an interpreter session.

tulipshell.py

# -*- coding: utf-8 -*-

"""Set up an interactive interpreter to have a underlying tulip loop.

The setup() function replaces the default loop policy, that starts a loop only
for the main thread, by a policy that has a single global loop available in any
thread.

The loop will be started when calling setup() in a background thread so that
the main thread could still be used by the interactive shell.

Main purpose of this module is to have a interactive Python shell with a
underlying tulip loop to test tasks w/o calling loop.run_until_complete() over
and over again (... and to get a hands-on feeling how tasks behave in a
looping environment).

Usage example:

    $ python
    >>> import tulipshell; tulipshell.setup()
    >>> future = tulipshell.testtask()  # task is executed in background
    >>> # get a cup of coffee or do some useful work
    >>> future.done()
    True
    >>> future.result().status
    200

testtask used in the example above (which is also included in this module) just
sends a GET request to http://google.com using tulip.http.request().

To exit the interpreter you need to stop the loop first, e.g. by pressing
Ctrl-C. Ctrl-D/exit() w/o stopping the loop doesn't work currently.
"""

# TODO(andi): find a way to stop the loop when interpreter receives EOF (Ctrl+D)

import os
import signal
import sys
import tempfile
import threading

import tulip
import tulip.http


class GlobalEventLoopPolicy(tulip.events.AbstractEventLoopPolicy):
    """Provides a global loop instance.

    Most of the code is copied from DefaultEventLoopPolicy, only the
    implementation of get_event_loop differs.

    Use with caution: this policy is intendend to be used when testing code
    that needs a loop in an interactive shell. It's not tested for thread-
    safety.
    """

    _loop = None

    def get_event_loop(self):
        """Get the event loop.

        This may be None or an instance of EventLoop.
        """
        if self._loop is None:
            self._loop = self.new_event_loop()
        return self._loop

    def set_event_loop(self, loop):
        """Set the event loop."""
        # TODO: The isinstance() test violates the PEP.
        assert loop is None or isinstance(loop, tulip.events.AbstractEventLoop)
        self._loop = loop

    def new_event_loop(self):
        """Create a new event loop.

        You must call set_event_loop() to make this the current event
        loop.
        """
        if sys.platform == 'win32':  # pragma: no cover
            from tulip import windows_events
            return windows_events.SelectorEventLoop()
        else:  # pragma: no cover
            from tulip import unix_events
            return unix_events.SelectorEventLoop()


def setup():
    """Configures tulip and starts the loop in a background thread."""
    tulip.set_event_loop_policy(GlobalEventLoopPolicy())
    loop = tulip.get_event_loop()
    loop.add_signal_handler(signal.SIGINT, loop.stop)

    # Add a dummy pipe to trigger selector events. Otherwise when tasks are
    # added after run_forever() was called the tasks won't be executed in most
    # cases since the selector doesn't receive any event.
    ping_read, ping_write = os.pipe()
    loop.add_reader(ping_read, lambda: None)
    f = open(ping_write, 'w') # open file, the fd is not valid in other threads
    @tulip.task
    def heartbeat():
        while True:
            yield from tulip.sleep(.2)
            f.write('')
    heartbeat_task = heartbeat()

    thread = threading.Thread(target=loop.run_forever)
    thread.start()
    # TODO(andi) shutdown properly


@tulip.task
def testtask():
    resp = yield from tulip.http.request('GET', 'http://google.com')
    return resp