vibefpmgui.dap

  1"""A Debug Adapter Protocol client: one debug adapter (gdb's ``--interpreter=dap``,
  2``lldb-dap``…) run as a process, spoken to over its stdin and stdout (see
  3`vibefpmgui.wire` for the framing; `vibefpmgui.debug_session` drives it).
  4
  5Each request gets its answer through a callback, ``callback(success, body,
  6message)``, called once: with the adapter's answer, or, when it doesn't
  7answer in time or stops, with success False and why. Events come as the
  8`event` signal. A request from the adapter (``runInTerminal``…) is answered
  9"not supported".
 10
 11As for fortls (`vibefpmgui.lsp`): no threads (the process is read in the GUI
 12thread), and methods, not lambdas, connected to the process.
 13"""
 14
 15from __future__ import annotations
 16
 17from collections.abc import Callable
 18
 19from qtpy.QtCore import QObject, QProcess, QProcessEnvironment, QTimer, Signal
 20
 21from .wire import FrameReader, frame
 22
 23TIMEOUT = 10_000  # ms a request may take (a hung adapter can't leave a session stuck)
 24
 25Callback = Callable[[bool, dict, str], None]
 26
 27
 28class DapClient(QObject):
 29    event = Signal(str, object)  # an event's name and body ({} if none)
 30    stopped = Signal(str)  # the adapter's process ended (why), not by stop()
 31
 32    def __init__(self, parent=None):
 33        super().__init__(parent)
 34        self.process: QProcess | None = None
 35        self._reader = FrameReader()
 36        self._seq = 0
 37        self._pending: dict[int, tuple[str, Callback, QTimer]] = {}
 38        self.capabilities: dict = {}
 39
 40    # ---------------------------------------------------------- lifecycle
 41    def is_running(self) -> bool:
 42        return self.process is not None
 43
 44    def start(self, command: list[str], cwd: str, env: dict[str, str] | None = None) -> bool:
 45        """Start the adapter (``command``: its program and arguments)."""
 46        self.stop()
 47        process = QProcess(self)
 48        process.setWorkingDirectory(cwd)
 49        if env:
 50            environment = QProcessEnvironment.systemEnvironment()
 51            for key, value in env.items():
 52                environment.insert(key, value)
 53            process.setProcessEnvironment(environment)
 54        process.setStandardErrorFile(QProcess.nullDevice())
 55        process.readyReadStandardOutput.connect(self._read)
 56        process.finished.connect(self._process_finished)
 57        process.errorOccurred.connect(self._process_error)
 58        self.process = process
 59        process.start(command[0], command[1:])
 60        return True
 61
 62    def stop(self):
 63        """End the adapter (and with it the program it debugs), at once."""
 64        process, self.process = self.process, None
 65        self._fail_pending("the debugger was stopped")
 66        self._reader.clear()
 67        if process is not None:
 68            process.kill()
 69            process.waitForFinished(1000)
 70            process.deleteLater()
 71
 72    def _process_finished(self, *_):
 73        if isinstance(self.sender(), QProcess):
 74            self._ended(self.sender())
 75
 76    def _process_error(self, error):
 77        if error == QProcess.FailedToStart and isinstance(self.sender(), QProcess):
 78            self._ended(self.sender())
 79
 80    def _ended(self, process: QProcess):
 81        if process is not self.process:
 82            return  # one we stopped
 83        self.process = None
 84        why = ("the debugger could not be started" if process.error() == QProcess.FailedToStart
 85               else f"the debugger stopped (exit code {process.exitCode()})")
 86        self._fail_pending(why)
 87        try:
 88            process.deleteLater()
 89        except RuntimeError:
 90            return  # the app is shutting down
 91        self.stopped.emit(why)
 92
 93    # ---------------------------------------------------------- messages
 94    def request(self, command: str, arguments: dict | None = None, callback: Callback | None = None,
 95                timeout: int = TIMEOUT) -> bool:
 96        """Send a request; False if no adapter is running (``callback`` is
 97        then called at once, with success False)."""
 98        if self.process is None:
 99            if callback is not None:
100                callback(False, {}, "no debugger is running")
101            return False
102        self._seq += 1
103        seq = self._seq
104        timer = QTimer(self)
105        timer.setSingleShot(True)
106        timer.setInterval(timeout)
107        timer.timeout.connect(self._timed_out)
108        timer.setProperty("seq", seq)
109        self._pending[seq] = (command, callback or _ignore, timer)
110        if timeout > 0:
111            timer.start()
112        self.process.write(frame({"seq": seq, "type": "request", "command": command,
113                                  "arguments": arguments or {}}))
114        return True
115
116    def _timed_out(self):
117        seq = self.sender().property("seq") if self.sender() is not None else None
118        entry = self._pending.pop(seq, None)
119        if entry is not None:
120            command, callback, timer = entry
121            timer.deleteLater()
122            _call(callback, False, {}, f"the debugger didn't answer {command} in time")
123
124    def _fail_pending(self, why: str):
125        pending, self._pending = self._pending, {}
126        for _command, callback, timer in pending.values():
127            timer.stop()
128            timer.deleteLater()
129            _call(callback, False, {}, why)
130
131    def _read(self):
132        if self.process is None:
133            return
134        for message in self._reader.feed(bytes(self.process.readAllStandardOutput())):
135            self._dispatch(message)
136
137    def _dispatch(self, message: dict):
138        kind = message.get("type")
139        if kind == "response":
140            entry = self._pending.pop(message.get("request_seq"), None)
141            if entry is None:
142                return
143            command, callback, timer = entry
144            timer.stop()
145            timer.deleteLater()
146            if command == "initialize" and message.get("success"):
147                self.capabilities = message.get("body") or {}
148            _call(callback, bool(message.get("success")), message.get("body") or {},
149                  str(message.get("message") or ""))
150        elif kind == "event":
151            self.event.emit(str(message.get("event") or ""), message.get("body") or {})
152        elif kind == "request" and self.process is not None:  # (runInTerminal…: not supported yet)
153            self._seq += 1
154            self.process.write(frame({"seq": self._seq, "type": "response", "request_seq": message.get("seq"),
155                                      "command": message.get("command"), "success": False,
156                                      "message": "not supported"}))
157
158
159def _ignore(*_):
160    pass
161
162
163def _call(callback: Callback, success: bool, body: dict, message: str):
164    try:
165        callback(success, body, message)
166    except RuntimeError:
167        pass  # what asked has been deleted (a closed panel)