| 26 | |
| 27 | |
| 28 | class Waiter: |
| 29 | def __init__(self, channel_owner: ChannelOwner, event: str) -> None: |
| 30 | self._result: asyncio.Future = asyncio.Future() |
| 31 | self._wait_id = uuid.uuid4().hex |
| 32 | self._loop = channel_owner._loop |
| 33 | self._pending_tasks: List[Task] = [] |
| 34 | self._channel = channel_owner._channel |
| 35 | self._registered_listeners: List[Tuple[EventEmitter, str, Callable]] = [] |
| 36 | self._logs: List[str] = [] |
| 37 | self._wait_for_event_info_before(self._wait_id, event) |
| 38 | |
| 39 | def _wait_for_event_info_before(self, wait_id: str, event: str) -> None: |
| 40 | self._channel.send_no_reply( |
| 41 | "__waitInfo__", |
| 42 | None, |
| 43 | { |
| 44 | "waitId": wait_id, |
| 45 | "phase": "before", |
| 46 | "event": event, |
| 47 | }, |
| 48 | title=f'Wait for event "{event}"', |
| 49 | ) |
| 50 | |
| 51 | def _wait_for_event_info_after(self, wait_id: str, error: Exception = None) -> None: |
| 52 | self._channel._connection.wrap_api_call_sync( |
| 53 | lambda: self._channel.send_no_reply( |
| 54 | "__waitInfo__", |
| 55 | None, |
| 56 | { |
| 57 | "waitId": wait_id, |
| 58 | "phase": "after", |
| 59 | **({"error": str(error)} if error else {}), |
| 60 | }, |
| 61 | ), |
| 62 | True, |
| 63 | ) |
| 64 | |
| 65 | def reject_on_event( |
| 66 | self, |
| 67 | emitter: EventEmitter, |
| 68 | event: str, |
| 69 | error: Union[Error, Callable[..., Error]], |
| 70 | predicate: Callable = None, |
| 71 | ) -> None: |
| 72 | def on_match() -> None: |
| 73 | self._reject(error() if callable(error) else error) |
| 74 | |
| 75 | def listener(event_data: Any = None) -> None: |
| 76 | self._evaluate_predicate(predicate, event_data, on_match) |
| 77 | |
| 78 | emitter.on(event, listener) |
| 79 | self._registered_listeners.append((emitter, event, listener)) |
| 80 | |
| 81 | def reject_on_timeout(self, timeout: float, message: str) -> None: |
| 82 | if timeout == 0: |
| 83 | return |
| 84 | |
| 85 | async def reject() -> None: |
no outgoing calls
no test coverage detected