Owns one persistent ``cmdop-core`` subprocess and the id-mux over it.
| 85 | async def _ensure_started(self) -> None: |
| 86 | if self._process is not None: |
| 87 | return |
| 88 | async with self._start_lock: |
| 89 | if self._process is not None: |
| 90 | return |
| 91 | env = dict(os.environ) |
| 92 | if self._config.token: |
| 93 | env["CMDOP_TOKEN"] = self._config.token |
| 94 | else: |
| 95 | env.pop("CMDOP_TOKEN", None) |
| 96 | if self._config.base_url: |
| 97 | env["CMDOP_BASE_URL"] = self._config.base_url |
| 98 | else: |
| 99 | env.pop("CMDOP_BASE_URL", None) |
| 100 | self._process = await asyncio.create_subprocess_exec( |
| 101 | self._binary_path, |
| 102 | "--stdio", |
| 103 | stdin=asyncio.subprocess.PIPE, |
| 104 | stdout=asyncio.subprocess.PIPE, |
| 105 | stderr=asyncio.subprocess.PIPE, |
| 106 | env=env, |
| 107 | ) |
| 108 | self._reader_task = asyncio.create_task(self._read_loop()) |
| 109 | self._stderr_task = asyncio.create_task(self._drain_stderr()) |
| 110 | |
| 111 | async def _drain_stderr(self) -> None: |
| 112 | assert self._process is not None and self._process.stderr is not None |
| 113 | while await self._process.stderr.readline(): |
| 114 | pass |
| 115 | |
| 116 | async def _read_loop(self) -> None: |
| 117 | assert self._process is not None and self._process.stdout is not None |
| 118 | try: |
| 119 | while True: |
| 120 | envelope = pb.Envelope.FromString( |
| 121 | await _read_delimited(self._process.stdout) |
| 122 | ) |
| 123 | self._dispatch(envelope) |
| 124 | except asyncio.IncompleteReadError: |
| 125 | self._fail_all(CmdopConnectionError("cmdop-core exited", code="transport")) |
| 126 | except asyncio.CancelledError: |
| 127 | raise |
| 128 | except Exception as exc: # noqa: BLE001 |
| 129 | self._fail_all( |
| 130 | CmdopConnectionError(f"core read failed: {exc}", code="transport") |
| 131 | ) |
| 132 | |
| 133 | def _dispatch(self, envelope: pb.Envelope) -> None: |
| 134 | pending = self._pending.get(envelope.id) |
| 135 | if pending is None: |
| 136 | return |
| 137 | if isinstance(pending, _UnaryPending): |
| 138 | if envelope.kind == pb.Envelope.KIND_ERROR: |
| 139 | self._pending.pop(envelope.id, None) |
| 140 | if not pending.future.done(): |
| 141 | pending.future.set_exception(map_core_error(envelope.error)) |
| 142 | elif envelope.kind == pb.Envelope.KIND_RESPONSE: |
| 143 | self._pending.pop(envelope.id, None) |
| 144 | if not pending.future.done(): |
no outgoing calls