| 125 | self._registered_listeners.append((emitter, event, listener)) |
| 126 | |
| 127 | def _evaluate_predicate( |
| 128 | self, |
| 129 | predicate: Optional[Callable], |
| 130 | event_data: Any, |
| 131 | on_match: Callable[[], None], |
| 132 | ) -> None: |
| 133 | if predicate is None: |
| 134 | on_match() |
| 135 | return |
| 136 | try: |
| 137 | result = predicate(event_data) |
| 138 | except Exception as e: |
| 139 | self._reject(e) |
| 140 | return |
| 141 | if inspect.iscoroutine(result): |
| 142 | |
| 143 | async def _await_predicate(coro: Any) -> None: |
| 144 | try: |
| 145 | matched = await coro |
| 146 | except Exception as e: |
| 147 | self._reject(e) |
| 148 | return |
| 149 | if matched and not self._result.done(): |
| 150 | on_match() |
| 151 | |
| 152 | self._pending_tasks.append(self._loop.create_task(_await_predicate(result))) |
| 153 | return |
| 154 | if result: |
| 155 | on_match() |
| 156 | |
| 157 | def result(self) -> asyncio.Future: |
| 158 | return self._result |