Close stdin (graceful drain), reap the child, cancel the read loop.
(self)
| 218 | self, request: pb.Envelope |
| 219 | ) -> tuple[int, asyncio.Queue[object]]: |
| 220 | request.id = self._allocate_id() |
| 221 | request.kind = pb.Envelope.KIND_REQUEST |
| 222 | queue: asyncio.Queue[object] = asyncio.Queue(maxsize=_STREAM_QUEUE_SIZE) |
| 223 | self._pending[request.id] = _StreamPending(queue) |
| 224 | try: |
| 225 | await self._write(request) |
| 226 | except BaseException: |
| 227 | self._pending.pop(request.id, None) |
| 228 | raise |
| 229 | return request.id, queue |
| 230 | |
| 231 | def _abandon_stream(self, call_id: int) -> None: |
| 232 | self._pending.pop(call_id, None) |
| 233 | |
| 234 | async def aclose(self) -> None: |
| 235 | if self._closed: |
| 236 | return |
| 237 | self._closed = True |
| 238 | process = self._process |
| 239 | if process is None: |
| 240 | return |
| 241 | if process.stdin is not None: |
| 242 | process.stdin.close() |
| 243 | try: |
| 244 | await asyncio.wait_for(process.wait(), 5) |
| 245 | except (TimeoutError, asyncio.TimeoutError): |
| 246 | process.kill() |
| 247 | await process.wait() |
| 248 | for task in (self._reader_task, self._stderr_task): |