Send a JSON-RPC message with a Content-Length header.
(self, message: dict)
| 239 | self.request_handlers[method] = handler |
| 240 | |
| 241 | async def _send_message(self, message: dict): |
| 242 | """Send a JSON-RPC message with a Content-Length header.""" |
| 243 | loop = self._loop or asyncio.get_event_loop() |
| 244 | |
| 245 | def write(): |
| 246 | content = json.dumps(message, separators=(",", ":")) |
| 247 | content_bytes = content.encode("utf-8") |
| 248 | header = f"Content-Length: {len(content_bytes)}\r\n\r\n" |
| 249 | with self._write_lock: |
| 250 | self.process.stdin.write(header.encode("utf-8")) |
| 251 | self.process.stdin.write(content_bytes) |
| 252 | self.process.stdin.flush() |
| 253 | |
| 254 | # Run in thread pool to avoid blocking |
| 255 | await loop.run_in_executor(None, write) |
| 256 | |
| 257 | def _read_loop(self): |
| 258 | """Read messages from the stream (runs in thread)""" |
no outgoing calls
no test coverage detected