/usr/lib/python2.7/dist-packages/prompt_toolkit/eventloop/inputhook.py is in python-prompt-toolkit 1.0.15-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """
Similar to `PyOS_InputHook` of the Python API. Some eventloops can have an
inputhook to allow easy integration with other event loops.
When the eventloop of prompt-toolkit is idle, it can call such a hook. This
hook can call another eventloop that runs for a short while, for instance to
keep a graphical user interface responsive.
It's the responsibility of this hook to exit when there is input ready.
There are two ways to detect when input is ready:
- Call the `input_is_ready` method periodically. Quit when this returns `True`.
- Add the `fileno` as a watch to the external eventloop. Quit when file descriptor
becomes readable. (But don't read from it.)
Note that this is not the same as checking for `sys.stdin.fileno()`. The
eventloop of prompt-toolkit allows thread-based executors, for example for
asynchronous autocompletion. When the completion for instance is ready, we
also want prompt-toolkit to gain control again in order to display that.
An alternative to using input hooks, is to create a custom `EventLoop` class that
controls everything.
"""
from __future__ import unicode_literals
import os
import threading
from prompt_toolkit.utils import is_windows
from .select import select_fds
__all__ = (
'InputHookContext',
)
class InputHookContext(object):
"""
Given as a parameter to the inputhook.
"""
def __init__(self, inputhook):
assert callable(inputhook)
self.inputhook = inputhook
self._input_is_ready = None
self._r, self._w = os.pipe()
def input_is_ready(self):
"""
Return True when the input is ready.
"""
return self._input_is_ready(wait=False)
def fileno(self):
"""
File descriptor that will become ready when the event loop needs to go on.
"""
return self._r
def call_inputhook(self, input_is_ready_func):
"""
Call the inputhook. (Called by a prompt-toolkit eventloop.)
"""
self._input_is_ready = input_is_ready_func
# Start thread that activates this pipe when there is input to process.
def thread():
input_is_ready_func(wait=True)
os.write(self._w, b'x')
threading.Thread(target=thread).start()
# Call inputhook.
self.inputhook(self)
# Flush the read end of the pipe.
try:
# Before calling 'os.read', call select.select. This is required
# when the gevent monkey patch has been applied. 'os.read' is never
# monkey patched and won't be cooperative, so that would block all
# other select() calls otherwise.
# See: http://www.gevent.org/gevent.os.html
# Note: On Windows, this is apparently not an issue.
# However, if we would ever want to add a select call, it
# should use `windll.kernel32.WaitForMultipleObjects`,
# because `select.select` can't wait for a pipe on Windows.
if not is_windows():
select_fds([self._r], timeout=None)
os.read(self._r, 1024)
except OSError:
# This happens when the window resizes and a SIGWINCH was received.
# We get 'Error: [Errno 4] Interrupted system call'
# Just ignore.
pass
self._input_is_ready = None
def close(self):
"""
Clean up resources.
"""
if self._r:
os.close(self._r)
os.close(self._w)
self._r = self._w = None
|