Read one JSON-RPC message from the LSP stdout.
(stream)
| 125 | |
| 126 | |
| 127 | def _read_message(stream) -> dict[str, Any]: |
| 128 | """Read one JSON-RPC message from the LSP stdout.""" |
| 129 | headers = _read_headers(stream) |
| 130 | length = int(headers["content-length"]) |
| 131 | body = b"" |
| 132 | while len(body) < length: |
| 133 | chunk = stream.read(length - len(body)) |
| 134 | if not chunk: |
| 135 | raise EOFError("LSP process closed stdout") |
| 136 | body += chunk |
| 137 | return json.loads(body) |
| 138 | |
| 139 | |
| 140 | # ── LspClient ──────────────────────────────────────────────────────────────── |
no test coverage detected