Internal: register the stream id + send its REQUEST. Called lazily by :class:`FrameStream` on first iteration so the same stream object can be re-iterated would-be-once semantics.
(self, req: pb.Envelope)
| 289 | return FrameStream(self, req) |
| 290 | |
| 291 | async def _start_stream(self, req: pb.Envelope) -> tuple[int, asyncio.Queue]: |
| 292 | """Internal: register the stream id + send its REQUEST. Called lazily by |
| 293 | :class:`FrameStream` on first iteration so the same stream object can be |
| 294 | re-iterated would-be-once semantics.""" |
| 295 | await self._ensure_started() |
| 296 | req.id = self._alloc_id() |
| 297 | req.kind = pb.Envelope.KIND_REQUEST |
| 298 | queue: asyncio.Queue = asyncio.Queue() |
| 299 | self._pending[req.id] = _StreamPending(queue) |
| 300 | self._write(req) |
| 301 | assert self._proc is not None and self._proc.stdin is not None |
| 302 | await self._proc.stdin.drain() |
| 303 | return req.id, queue |
no test coverage detected