| 46 | await self._stopped_future |
| 47 | |
| 48 | async def connect(self) -> None: |
| 49 | self._stopped_future: asyncio.Future = asyncio.Future() |
| 50 | |
| 51 | def handle_message(message: Dict) -> None: |
| 52 | if self._stop_requested: |
| 53 | return |
| 54 | self.on_message(cast(ParsedMessagePayload, message)) |
| 55 | |
| 56 | def handle_closed(reason: Optional[str]) -> None: |
| 57 | self.emit("close", reason) |
| 58 | if reason: |
| 59 | self.on_error_future.set_exception(TargetClosedError(reason)) |
| 60 | self._stopped_future.set_result(None) |
| 61 | |
| 62 | self._pipe_channel.on( |
| 63 | "message", |
| 64 | lambda params: handle_message(params["message"]), |
| 65 | ) |
| 66 | self._pipe_channel.on( |
| 67 | "closed", |
| 68 | lambda params: handle_closed(params.get("reason")), |
| 69 | ) |
| 70 | |
| 71 | async def run(self) -> None: |
| 72 | await self._stopped_future |