Add pending() and flush() to BaseEventQueue

Author: tanloong

Lib/_pyrepl/base_eventqueue.py

@@ -54,6 +54,51 @@ def empty(self) -> bool:
         """
         return not self.events
 
+    def pending(self) -> bool:
+        """
+        True when we have received the start of an escape/key sequence but
+        are still waiting for further bytes to disambiguate it.
+
+        This is the case right after a byte (such as ESC) that our keymap
+        knows only as a prefix of a longer sequence has been pushed, but
+        before the rest of that sequence arrives.  Consoles use this to
+        decide whether the next read should block indefinitely or only for
+        the escape timeout (see ``flush``).
+        """
+        return self.keymap is not self.compiled_keymap
+
+    def flush(self) -> None:
+        """
+        Finalize any pending, incomplete input as discrete events.
+
+        This implements the classic "escape-time" disambiguation used by
+        TUIs such as ncurses, Vim and Tmux: when we have received the start
+        of an escape sequence (e.g. a lone ESC) but no further bytes arrive
+        within the escape timeout, we stop waiting and emit the bytes
+        accumulated so far as a complete key instead.
+
+        The behavior mirrors that of an unrecognized escape sequence: the
+        leading ESC is emitted as a key and any remaining bytes are fed
+        back through ``push`` so they can start another sequence or be
+        treated as ordinary text.  Calling this when there is nothing
+        pending is a no-op.
+        """
+        if self.keymap is self.compiled_keymap:
+            # Nothing pending: either idle, or a complete key was already
+            # emitted by ``push``.
+            return
+        buf = self.buf.take_bytes()
+        self.keymap = self.compiled_keymap
+        if buf and buf[0] == 27:  # escape
+            self.insert(Event('key', '\033', b'\033'))
+            for _c in buf[1:]:
+                self.push(_c)
+        else:
+            # Defensive: a pending prefix that does not start with ESC
+            # (e.g. a partial multi-byte sequence). Emit it as text.
+            data = bytes(buf).decode(self.encoding, 'replace')
+            self.insert(Event('key', data, buf))
+
     def insert(self, event: Event) -> None:
         """
         Inserts an event into the queue.