Add wait_for_event helper to CDPConnection

Co-authored-by: P4x-ng <P4X-ng@users.noreply.github.com>

Author: cursoragent

README.md

@@ -57,10 +57,14 @@ async def main():
     # Connect to a Chrome DevTools Protocol endpoint
     async with CDPConnection("ws://localhost:9222/devtools/page/YOUR_PAGE_ID") as conn:
         # Navigate to a URL
-        frame_id, loader_id, error = await conn.execute(
+        frame_id, loader_id, error_text, is_download = await conn.execute(
             page.navigate(url="https://example.com")
         )
-        print(f"Navigated to example.com, frame_id: {frame_id}")
+        print(f"Navigated to example.com, frame_id: {frame_id}, is_download={is_download}")
+
+        # Wait for a specific event (new helper API)
+        load_event = await conn.wait_for_event(page.LoadEventFired, timeout=10.0)
+        print(f"Load event timestamp: {load_event.timestamp}")
 
 asyncio.run(main())
 ```
@@ -71,6 +75,7 @@ asyncio.run(main())
 - **JSON-RPC Framing**: Automatic message ID assignment and request/response matching
 - **Command Multiplexing**: Execute multiple commands concurrently with proper tracking
 - **Event Handling**: Async iterator for receiving browser events
+- **Event Waiting**: Wait for a specific event type (optionally with predicates)
 - **Error Handling**: Comprehensive error handling with typed exceptions
 
 See the [examples directory](examples/) for more usage patterns.
@@ -94,6 +99,23 @@ For detailed API documentation, see:
 - **[Chrome DevTools Protocol](https://chromedevtools.github.io/devtools-protocol/)** - Official CDP specification
 - **[Examples](examples/)** - Code examples demonstrating usage patterns
 
+### Event waiting helper
+
+`CDPConnection.wait_for_event()` can simplify flows where you need to wait for one
+specific event:
+
+```python
+await conn.execute(page.enable())
+await conn.execute(page.navigate(url="https://example.com"))
+
+load_event = await conn.wait_for_event(
+    page.LoadEventFired,
+    predicate=lambda evt: evt.timestamp > 0,
+    timeout=10.0,
+)
+print(load_event.timestamp)
+```
+
 ### Key Modules
 
 - `cdp.connection` - WebSocket I/O and connection management (I/O mode)
@@ -119,41 +141,4 @@ For information about reporting security vulnerabilities, please see our [Securi
 
 This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
 
-## API Reference
-
-The library provides Python wrappers for all Chrome DevTools Protocol domains:
-
-- **Page**: Page control (navigation, screenshots, etc.)
-- **DOM**: DOM inspection and manipulation
-- **Network**: Network monitoring and interception
-- **Runtime**: JavaScript execution and evaluation
-- **Debugger**: JavaScript debugging
-- **Performance**: Performance metrics and profiling
-- **Security**: Security-related information
-- And many more...
-
-For complete API documentation, visit [py-cdp.readthedocs.io](https://py-cdp.readthedocs.io).
-
-### Type System
-
-All CDP types, commands, and events are fully typed with Python type hints, providing:
-- IDE autocomplete support
-- Static type checking with mypy
-- Clear API contracts
-- Inline documentation
-
-## Contributing
-
-We welcome contributions! Please see our [Contributing Guide](CONTRIBUTING.md) for details on:
-- How to report bugs and request features
-- Development setup and workflow
-- Coding standards and testing requirements
-- Pull request process
-
-For questions or discussions, feel free to open an issue on GitHub.
-
-## License
-
-This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
-
 <a href="https://www.hyperiongray.com/?pk_campaign=github&pk_kwd=pycdp"><img alt="define hyperion gray" width="500px" src="https://hyperiongray.s3.amazonaws.com/define-hg.svg"></a>

cdp/connection.py

@@ -11,7 +11,7 @@
 import json
 import logging
 import typing
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 
 try:
     import websockets
@@ -55,6 +55,22 @@ class PendingCommand:
     params: T_JSON_DICT
 
 
+@dataclass
+class EventWaiter:
+    """Represents a consumer waiting for a matching event."""
+    future: asyncio.Future
+    event_type: typing.Optional[typing.Union[type, typing.Tuple[type, ...]]] = None
+    predicate: typing.Optional[typing.Callable[[typing.Any], bool]] = None
+
+    def matches(self, event: typing.Any) -> bool:
+        """Return True if this waiter should receive the event."""
+        if self.event_type is not None and not isinstance(event, self.event_type):
+            return False
+        if self.predicate is not None and not self.predicate(event):
+            return False
+        return True
+
+
 class CDPConnection:
     """
     Manages a WebSocket connection to Chrome DevTools Protocol.
@@ -96,6 +112,7 @@ def __init__(self, url: str, timeout: float = 30.0):
         self._next_command_id = 1
         self._pending_commands: typing.Dict[int, PendingCommand] = {}
         self._event_queue: asyncio.Queue = asyncio.Queue()
+        self._event_waiters: typing.List[EventWaiter] = []
         self._recv_task: typing.Optional[asyncio.Task] = None
         self._closed = False
 
@@ -131,6 +148,12 @@ async def close(self) -> None:
             if not pending.future.done():
                 pending.future.cancel()
         self._pending_commands.clear()
+
+        # Cancel all event waiters
+        for waiter in self._event_waiters:
+            if not waiter.future.done():
+                waiter.future.set_exception(CDPConnectionError("Connection closed"))
+        self._event_waiters.clear()
 
         # Close the WebSocket
         if self._ws:
@@ -213,9 +236,28 @@ async def _handle_event(self, data: T_JSON_DICT) -> None:
         """Handle an event notification."""
         try:
             event = parse_json_event(data)
+            self._notify_event_waiters(event)
             await self._event_queue.put(event)
         except Exception as e:
             logger.error(f"Failed to parse event: {e}")
+
+    def _notify_event_waiters(self, event: typing.Any) -> None:
+        """Resolve any pending waiters that match the event."""
+        if not self._event_waiters:
+            return
+
+        remaining_waiters: typing.List[EventWaiter] = []
+        for waiter in self._event_waiters:
+            if waiter.future.done():
+                continue
+            try:
+                if waiter.matches(event):
+                    waiter.future.set_result(event)
+                else:
+                    remaining_waiters.append(waiter)
+            except Exception as e:
+                waiter.future.set_exception(e)
+        self._event_waiters = remaining_waiters
 
     async def execute(
         self,
@@ -330,6 +372,48 @@ def get_event_nowait(self) -> typing.Optional[typing.Any]:
             return self._event_queue.get_nowait()
         except asyncio.QueueEmpty:
             return None
+
+    async def wait_for_event(
+        self,
+        event_type: typing.Optional[typing.Union[type, typing.Tuple[type, ...]]] = None,
+        predicate: typing.Optional[typing.Callable[[typing.Any], bool]] = None,
+        timeout: typing.Optional[float] = None,
+    ) -> typing.Any:
+        """
+        Wait for the next event matching the provided filters.
+
+        Args:
+            event_type: Optional event class (or tuple of classes) to match.
+            predicate: Optional callable that must return True for a match.
+            timeout: Optional timeout override in seconds.
+
+        Returns:
+            The matching CDP event object.
+
+        Raises:
+            CDPConnectionError: If the connection is closed.
+            asyncio.TimeoutError: If no matching event arrives in time.
+        """
+        if self._closed:
+            raise CDPConnectionError("Connection closed")
+        if self._ws is None:
+            raise CDPConnectionError("Not connected")
+
+        future: asyncio.Future = asyncio.Future()
+        waiter = EventWaiter(future=future, event_type=event_type, predicate=predicate)
+        self._event_waiters.append(waiter)
+
+        timeout_val = timeout if timeout is not None else self.timeout
+        try:
+            return await asyncio.wait_for(future, timeout=timeout_val)
+        except asyncio.TimeoutError:
+            if waiter in self._event_waiters:
+                self._event_waiters.remove(waiter)
+            raise asyncio.TimeoutError("Timed out waiting for matching CDP event")
+        except Exception:
+            if waiter in self._event_waiters:
+                self._event_waiters.remove(waiter)
+            raise
 
     @property
     def is_connected(self) -> bool:

docs/connection.md

@@ -21,16 +21,21 @@ async def main():
     # Connect using async context manager
     async with CDPConnection("ws://localhost:9222/devtools/page/YOUR_PAGE_ID") as conn:
         # Execute a command
-        frame_id, loader_id, error = await conn.execute(
+        frame_id, loader_id, error_text, is_download = await conn.execute(
             page.navigate(url="https://example.com")
         )
+        print(f"Navigated: frame_id={frame_id} is_download={is_download}")
 
         # Evaluate JavaScript
         result, exception = await conn.execute(
             runtime.evaluate(expression="document.title")
         )
         print(f"Page title: {result.value}")
 
+        # Wait for a specific event
+        event = await conn.wait_for_event(page.LoadEventFired, timeout=10.0)
+        print(f"Page loaded at {event.timestamp}")
+
 asyncio.run(main())
 ```
 
@@ -120,6 +125,20 @@ if event:
     print(f"Got event: {event}")
 ```
 
+Wait for a single matching event without manually iterating:
+
+```python
+# Wait for the next LoadEventFired
+event = await conn.wait_for_event(page.LoadEventFired, timeout=10.0)
+
+# Wait using a custom predicate
+event = await conn.wait_for_event(
+    page.LoadEventFired,
+    predicate=lambda evt: evt.timestamp > 0,
+    timeout=10.0,
+)
+```
+
 ### Error Handling
 
 The connection module provides typed exceptions:
@@ -162,6 +181,12 @@ class CDPConnection:
     async def execute(self, cmd, timeout: Optional[float] = None) -> Any
     async def listen(self) -> AsyncIterator[Any]
     def get_event_nowait(self) -> Optional[Any]
+    async def wait_for_event(
+        self,
+        event_type: Optional[Union[type, Tuple[type, ...]]] = None,
+        predicate: Optional[Callable[[Any], bool]] = None,
+        timeout: Optional[float] = None,
+    ) -> Any
 
     @property
     def is_connected(self) -> bool

examples/connection_example.py

@@ -6,7 +6,8 @@
 1. Connect to a Chrome DevTools Protocol endpoint
 2. Execute commands
 3. Handle events
-4. Multiplex multiple commands concurrently
+4. Wait for a specific event
+5. Multiplex multiple commands concurrently
 """
 
 import asyncio
@@ -76,6 +77,25 @@ async def event_handling_example():
         print("Navigation complete!")
 
 
+async def wait_for_event_example():
+    """Example showing wait_for_event helper usage."""
+    print("\n=== wait_for_event Example ===")
+
+    url = "ws://localhost:9222/devtools/page/YOUR_PAGE_ID"
+
+    async with CDPConnection(url) as conn:
+        await conn.execute(page.enable())
+        await conn.execute(page.navigate(url="https://example.com"))
+
+        # Wait for a matching load event instead of manually iterating.
+        load_event = await conn.wait_for_event(
+            page.LoadEventFired,
+            predicate=lambda evt: evt.timestamp > 0,
+            timeout=10.0,
+        )
+        print(f"Page loaded at timestamp: {load_event.timestamp}")
+
+
 async def multiplexing_example():
     """Example showing concurrent command execution (multiplexing)."""
     print("\n=== Multiplexing Example ===")
@@ -138,6 +158,7 @@ async def main():
     # Uncomment the examples you want to run:
     # await basic_example()
     # await event_handling_example()
+    # await wait_for_event_example()
     # await multiplexing_example()
     # await error_handling_example()