(self)
| 135 | self._output = self._proc.stdin |
| 136 | |
| 137 | async def run(self) -> None: |
| 138 | assert self._proc.stdout |
| 139 | assert self._proc.stdin |
| 140 | try: |
| 141 | while not self._stopped: |
| 142 | try: |
| 143 | buffer = await self._proc.stdout.readexactly(4) |
| 144 | if self._stopped: |
| 145 | break |
| 146 | length = int.from_bytes(buffer, byteorder="little", signed=False) |
| 147 | buffer = bytes(0) |
| 148 | while length: |
| 149 | to_read = min(length, 32768) |
| 150 | data = await self._proc.stdout.readexactly(to_read) |
| 151 | if self._stopped: |
| 152 | break |
| 153 | length -= to_read |
| 154 | if len(buffer): |
| 155 | buffer = buffer + data |
| 156 | else: |
| 157 | buffer = data |
| 158 | if self._stopped: |
| 159 | break |
| 160 | |
| 161 | obj = self.deserialize_message(buffer) |
| 162 | self.on_message(obj) |
| 163 | except asyncio.IncompleteReadError: |
| 164 | if not self._stopped: |
| 165 | self.on_error_future.set_exception( |
| 166 | Exception("Connection closed while reading from the driver") |
| 167 | ) |
| 168 | break |
| 169 | await asyncio.sleep(0) |
| 170 | |
| 171 | await self._proc.communicate() |
| 172 | finally: |
| 173 | # Release waiters on wait_until_stopped() even if this task was |
| 174 | # cancelled before reaching the end (e.g. by asyncio.run()'s |
| 175 | # task-cancellation phase that runs before asyncio-atexit hooks). |
| 176 | if not self._stopped_future.done(): |
| 177 | self._stopped_future.set_result(None) |
| 178 | |
| 179 | def send(self, message: Dict) -> None: |
| 180 | assert self._output |
nothing calls this directly
no test coverage detected