Read a JSON-RPC message from the LSP server
(self)
| 119 | logger.error(f'Error sending LSP notification: {e}') |
| 120 | |
| 121 | async def _read_message(self) -> dict: |
| 122 | """Read a JSON-RPC message from the LSP server""" |
| 123 | if not self.stdout: |
| 124 | raise RuntimeError('LSP server stdout not available') |
| 125 | |
| 126 | # Read headers |
| 127 | headers = {} |
| 128 | while True: |
| 129 | line = await self.stdout.readline() |
| 130 | if not line: |
| 131 | raise RuntimeError('LSP server closed connection') |
| 132 | line = line.decode('utf-8').strip() |
| 133 | if not line: |
| 134 | break |
| 135 | if ':' in line: |
| 136 | key, value = line.split(':', 1) |
| 137 | headers[key.strip()] = value.strip() |
| 138 | |
| 139 | # Read content |
| 140 | content_length = int(headers.get('Content-Length', 0)) |
| 141 | if content_length > 0: |
| 142 | content = await self.stdout.readexactly(content_length) |
| 143 | logger.info('LSP:' + content.decode('utf-8')) |
| 144 | return json.loads(content.decode('utf-8')) |
| 145 | return {} |
| 146 | |
| 147 | async def initialize(self): |
| 148 | """Initialize the LSP server and wait for it to be ready""" |
no test coverage detected