glass/browser/
cdp.rs

1//! Low-level Chrome DevTools Protocol (CDP) client.
2//!
3//! Provides a multiplexed WebSocket connection to a Chrome debugging
4//! endpoint, including command dispatch, event subscription, and
5//! session-scoped target management.
6
7use futures_util::{SinkExt, StreamExt};
8use serde::de::DeserializeOwned;
9use serde::{Deserialize, Serialize};
10use serde_json::Value;
11use std::collections::HashMap;
12use std::error::Error;
13use std::fmt::{Display, Formatter};
14use std::path::Path;
15use std::sync::{
16    Arc,
17    atomic::{AtomicU64, Ordering},
18};
19use std::time::Duration;
20use tokio::sync::{broadcast, mpsc, oneshot};
21use tokio_tungstenite::tungstenite::Message;
22use tracing::{debug, info, warn};
23
24#[derive(Clone, Default)]
25struct CdpRoute {
26    target_id: Option<String>,
27    session_id: Option<String>,
28    context_id: Option<i64>,
29    frame_id: Option<String>,
30}
31
32tokio::task_local! {
33    static OPERATION_ROUTE: CdpRoute;
34    static CDP_WAIT_SCOPE: Arc<AtomicU64>;
35}
36
37/// A CDP method call request.
38#[derive(Debug, Serialize)]
39pub struct CdpRequest {
40    pub id: u64,
41    pub method: String,
42    #[serde(skip_serializing_if = "Option::is_none")]
43    pub params: Option<Value>,
44    #[serde(rename = "sessionId", skip_serializing_if = "Option::is_none")]
45    pub session_id: Option<String>,
46}
47
48/// A protocol or transport error returned by a CDP connection.
49#[derive(Debug, Clone, Deserialize, Serialize)]
50pub struct CdpError {
51    pub code: i64,
52    pub message: String,
53    #[serde(default, skip_serializing_if = "Option::is_none")]
54    pub data: Option<Value>,
55    #[serde(skip)]
56    kind: CdpErrorKind,
57}
58
59#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
60enum CdpErrorKind {
61    #[default]
62    Protocol,
63    Transport,
64    ResponseTimeout,
65}
66
67impl CdpError {
68    fn transport(message: impl Into<String>) -> Self {
69        Self {
70            code: -32_000,
71            message: message.into(),
72            data: None,
73            kind: CdpErrorKind::Transport,
74        }
75    }
76
77    fn response_timeout(timeout: Duration) -> Self {
78        Self {
79            code: -32_000,
80            message: format!(
81                "CDP response timeout after {} seconds",
82                timeout.as_secs_f64()
83            ),
84            data: None,
85            kind: CdpErrorKind::ResponseTimeout,
86        }
87    }
88
89    fn decode(error: impl std::fmt::Display) -> Self {
90        Self::transport(format!("failed to decode typed CDP response: {error}"))
91    }
92
93    /// Whether this error is the locally typed expiry of an unanswered CDP request.
94    pub fn is_response_timeout(&self) -> bool {
95        self.kind == CdpErrorKind::ResponseTimeout
96    }
97}
98
99impl Display for CdpError {
100    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
101        write!(formatter, "CDP error {}: {}", self.code, self.message)
102    }
103}
104
105impl Error for CdpError {}
106
107/// Lightweight notification of a CDP event.
108///
109/// This default event stream intentionally omits payloads. Subscribe to
110/// [`CdpClient::subscribe_events_with_params`] only when a caller needs them.
111#[derive(Debug, Clone)]
112pub struct CdpEvent {
113    pub method: String,
114}
115
116/// CDP event with its full JSON payload for explicit diagnostic or media use.
117#[derive(Debug, Clone)]
118pub struct CdpEventWithParams {
119    pub method: String,
120    pub params: Value,
121    pub session_id: Option<String>,
122}
123
124#[derive(Debug)]
125pub struct CdpScreencastFrame {
126    pub data: String,
127    pub metadata: Value,
128    pub session_id: Option<String>,
129}
130
131struct ScreencastSink {
132    session_id: Option<String>,
133    sender: mpsc::Sender<CdpScreencastFrame>,
134}
135
136#[derive(Debug, Deserialize)]
137struct IncomingMessage {
138    #[serde(default)]
139    id: Option<u64>,
140    #[serde(default)]
141    result: Option<Value>,
142    #[serde(default)]
143    error: Option<CdpError>,
144    #[serde(default)]
145    method: Option<String>,
146    #[serde(default, rename = "sessionId")]
147    session_id: Option<String>,
148}
149
150#[derive(Debug, Deserialize)]
151struct IncomingEventParams {
152    #[serde(default)]
153    params: Value,
154}
155
156enum Command {
157    Request {
158        id: u64,
159        json: String,
160        response: oneshot::Sender<Result<Value, CdpError>>,
161    },
162    Cancel {
163        id: u64,
164    },
165    FireAndForget {
166        json: String,
167    },
168    Close,
169}
170
171struct PendingRequestGuard {
172    tx: mpsc::UnboundedSender<Command>,
173    id: u64,
174    armed: bool,
175}
176
177impl PendingRequestGuard {
178    fn disarm(&mut self) {
179        self.armed = false;
180    }
181}
182
183impl Drop for PendingRequestGuard {
184    fn drop(&mut self) {
185        if self.armed {
186            let _ = self.tx.send(Command::Cancel { id: self.id });
187        }
188    }
189}
190
191/// A multiplexed connection to Chrome DevTools Protocol.
192///
193/// The connection task owns the WebSocket and the pending request map. This
194/// keeps response routing and event delivery in one place while allowing
195/// multiple commands to be in flight at once.
196#[derive(Clone)]
197pub struct CdpClient {
198    tx: mpsc::UnboundedSender<Command>,
199    next_id: Arc<AtomicU64>,
200    events: broadcast::Sender<CdpEvent>,
201    payload_events: broadcast::Sender<CdpEventWithParams>,
202    screencast_sink: Arc<std::sync::Mutex<Option<ScreencastSink>>>,
203    screencast_received: Arc<AtomicU64>,
204    screencast_dropped: Arc<AtomicU64>,
205    cdp_wait_nanos: Arc<AtomicU64>,
206    timeout: Duration,
207    active_route: Arc<std::sync::Mutex<CdpRoute>>,
208}
209
210struct RemoteObjectBatchGuard {
211    cdp: CdpClient,
212    object_ids: Vec<String>,
213}
214
215impl RemoteObjectBatchGuard {
216    fn new(cdp: CdpClient, array_id: String) -> Self {
217        Self {
218            cdp,
219            object_ids: vec![array_id],
220        }
221    }
222
223    async fn cleanup(&mut self) -> Result<(), CdpError> {
224        let object_ids = std::mem::take(&mut self.object_ids);
225        let mut first_error = None;
226        for object_id in object_ids {
227            if let Err(error) = self.cdp.release_object(&object_id).await
228                && first_error.is_none()
229            {
230                first_error = Some(error);
231            }
232        }
233        first_error.map_or(Ok(()), Err)
234    }
235}
236
237impl Drop for RemoteObjectBatchGuard {
238    fn drop(&mut self) {
239        let cdp = self.cdp.clone();
240        let object_ids = std::mem::take(&mut self.object_ids);
241        tokio::spawn(async move {
242            for object_id in object_ids {
243                let _ = cdp.release_object(&object_id).await;
244            }
245        });
246    }
247}
248
249impl CdpClient {
250    /// Number of CDP requests allocated by this connection so far.
251    ///
252    /// This is a monotonic diagnostic counter and does not retain request
253    /// payloads or alter command routing.
254    pub fn request_count(&self) -> u64 {
255        self.next_id.load(Ordering::Relaxed).saturating_sub(1)
256    }
257
258    /// Total time spent awaiting CDP responses on this connection.
259    ///
260    /// This is a monotonic diagnostic counter used by the benchmark to
261    /// separate protocol wait time from Glass-side orchestration work.
262    pub fn cdp_wait_nanos(&self) -> u64 {
263        self.cdp_wait_nanos.load(Ordering::Relaxed)
264    }
265
266    /// Measure CDP response wait time initiated by one async operation.
267    ///
268    /// The scope is task-local, so detached cleanup work does not get charged
269    /// to the operation that caused it.
270    pub async fn measure_cdp_wait<F>(&self, future: F) -> (F::Output, u64)
271    where
272        F: std::future::Future,
273    {
274        let wait_nanos = Arc::new(AtomicU64::new(0));
275        let output = CDP_WAIT_SCOPE.scope(Arc::clone(&wait_nanos), future).await;
276        (output, wait_nanos.load(Ordering::Relaxed))
277    }
278
279    /// Connect to a Chrome CDP page WebSocket using the default timeout.
280    pub async fn connect(ws_url: &str) -> Result<Self, Box<dyn Error>> {
281        Self::connect_with_timeout(ws_url, Duration::from_secs(30)).await
282    }
283
284    /// Connect to a Chrome CDP page WebSocket with a custom command timeout.
285    pub async fn connect_with_timeout(
286        ws_url: &str,
287        timeout: Duration,
288    ) -> Result<Self, Box<dyn Error>> {
289        info!(%ws_url, "connecting to CDP");
290        let (ws_stream, _) = tokio_tungstenite::connect_async(ws_url).await?;
291        let (mut write, mut read) = ws_stream.split();
292        let (tx, mut rx) = mpsc::unbounded_channel::<Command>();
293        let (event_tx, _) = broadcast::channel::<CdpEvent>(128);
294        let (payload_event_tx, _) = broadcast::channel::<CdpEventWithParams>(128);
295        let actor_events = event_tx.clone();
296        let actor_payload_events = payload_event_tx.clone();
297        let screencast_sink = Arc::new(std::sync::Mutex::new(None));
298        let actor_screencast_sink = Arc::clone(&screencast_sink);
299        let screencast_received = Arc::new(AtomicU64::new(0));
300        let actor_screencast_received = Arc::clone(&screencast_received);
301        let screencast_dropped = Arc::new(AtomicU64::new(0));
302        let actor_screencast_dropped = Arc::clone(&screencast_dropped);
303        let cdp_wait_nanos = Arc::new(AtomicU64::new(0));
304        let actor_tx = tx.clone();
305        let actor_next_id = Arc::new(AtomicU64::new(1));
306        let next_id = Arc::clone(&actor_next_id);
307
308        tokio::spawn(async move {
309            let mut pending: HashMap<u64, oneshot::Sender<Result<Value, CdpError>>> =
310                HashMap::new();
311            let mut close_reason = "CDP connection closed".to_string();
312
313            loop {
314                tokio::select! {
315                    command = rx.recv() => {
316                        match command {
317                            Some(Command::Request { id, json, response }) => {
318                                pending.insert(id, response);
319                                if let Err(error) = write.send(Message::Text(json.into())).await {
320                                    close_reason = format!("CDP write failed: {error}");
321                                    break;
322                                }
323                            }
324                            Some(Command::Cancel { id }) => {
325                                pending.remove(&id);
326                            }
327                            Some(Command::FireAndForget { json }) => {
328                                if let Err(error) = write.send(Message::Text(json.into())).await {
329                                    close_reason = format!("CDP write failed: {error}");
330                                    break;
331                                }
332                            }
333                            Some(Command::Close) | None => {
334                                let _ = write.send(Message::Close(None)).await;
335                                close_reason = "CDP connection closed by client".to_string();
336                                break;
337                            }
338                        }
339                    }
340                    message = read.next() => {
341                        match message {
342                            Some(Ok(Message::Text(text))) => {
343                                handle_incoming_message(
344                                    &mut pending,
345                                    &actor_events,
346                                    &actor_payload_events,
347                                    ScreencastDispatch {
348                                        sink: &actor_screencast_sink,
349                                        received: &actor_screencast_received,
350                                        dropped: &actor_screencast_dropped,
351                                        command_tx: &actor_tx,
352                                        next_id: &actor_next_id,
353                                    },
354                                    text.as_ref(),
355                                );
356                            }
357                            Some(Ok(Message::Binary(bytes))) => {
358                                match std::str::from_utf8(bytes.as_ref()) {
359                                    Ok(text) => handle_incoming_message(
360                                        &mut pending,
361                                        &actor_events,
362                                        &actor_payload_events,
363                                        ScreencastDispatch {
364                                            sink: &actor_screencast_sink,
365                                            received: &actor_screencast_received,
366                                            dropped: &actor_screencast_dropped,
367                                            command_tx: &actor_tx,
368                                            next_id: &actor_next_id,
369                                        },
370                                        text,
371                                    ),
372                                    Err(error) => warn!(%error, "ignoring non-UTF-8 CDP frame"),
373                                }
374                            }
375                            Some(Ok(Message::Ping(payload))) => {
376                                if let Err(error) = write.send(Message::Pong(payload)).await {
377                                    close_reason = format!("CDP pong failed: {error}");
378                                    break;
379                                }
380                            }
381                            Some(Ok(Message::Close(_))) => {
382                                close_reason = "CDP server closed the connection".to_string();
383                                break;
384                            }
385                            Some(Ok(_)) => {}
386                            Some(Err(error)) => {
387                                close_reason = format!("CDP read failed: {error}");
388                                break;
389                            }
390                            None => break,
391                        }
392                    }
393                }
394            }
395
396            let error = CdpError::transport(close_reason);
397            for (_, response) in pending.drain() {
398                let _ = response.send(Err(error.clone()));
399            }
400        });
401
402        Ok(Self {
403            tx,
404            next_id,
405            events: event_tx,
406            payload_events: payload_event_tx,
407            screencast_sink,
408            screencast_received,
409            screencast_dropped,
410            cdp_wait_nanos,
411            timeout,
412            active_route: Arc::new(std::sync::Mutex::new(CdpRoute::default())),
413        })
414    }
415
416    /// Subscribe to method-only CDP events such as page lifecycle events.
417    pub fn subscribe_events(&self) -> broadcast::Receiver<CdpEvent> {
418        self.events.subscribe()
419    }
420
421    /// Subscribe to CDP events with payloads for explicit media or diagnostic work.
422    ///
423    /// Payload JSON is parsed only while this stream has at least one receiver.
424    pub fn subscribe_events_with_params(&self) -> broadcast::Receiver<CdpEventWithParams> {
425        self.payload_events.subscribe()
426    }
427
428    pub fn open_screencast_channel(
429        &self,
430        session_id: Option<String>,
431    ) -> Result<mpsc::Receiver<CdpScreencastFrame>, CdpError> {
432        let mut sink = self
433            .screencast_sink
434            .lock()
435            .map_err(|_| CdpError::transport("screencast sink lock poisoned"))?;
436        if sink.is_some() {
437            return Err(CdpError::transport("a screencast scope is already active"));
438        }
439        let (sender, receiver) = mpsc::channel(2);
440        *sink = Some(ScreencastSink { session_id, sender });
441        self.screencast_received.store(0, Ordering::Relaxed);
442        self.screencast_dropped.store(0, Ordering::Relaxed);
443        Ok(receiver)
444    }
445
446    pub fn close_screencast_channel(&self) -> (u64, u64) {
447        if let Ok(mut sink) = self.screencast_sink.lock() {
448            *sink = None;
449        }
450        (
451            self.screencast_received.load(Ordering::Relaxed),
452            self.screencast_dropped.load(Ordering::Relaxed),
453        )
454    }
455
456    pub fn screencast_stats(&self) -> (u64, u64) {
457        (
458            self.screencast_received.load(Ordering::Relaxed),
459            self.screencast_dropped.load(Ordering::Relaxed),
460        )
461    }
462
463    pub fn current_session_id(&self) -> Option<String> {
464        self.current_route().session_id
465    }
466
467    pub async fn set_domain_enabled_for(
468        &self,
469        session_id: Option<String>,
470        domain: &str,
471        enabled: bool,
472    ) -> Result<(), CdpError> {
473        let method = format!("{domain}.{}", if enabled { "enable" } else { "disable" });
474        self.send_routed(&method, None, session_id, self.timeout)
475            .await?;
476        Ok(())
477    }
478
479    /// Send a CDP command and wait for the response.
480    pub async fn send(&self, method: &str, params: Option<Value>) -> Result<Value, CdpError> {
481        let session_id = self.current_route().session_id;
482        self.send_routed(method, params, session_id, self.timeout)
483            .await
484    }
485
486    /// Send a command and deserialize its response into a caller-owned type.
487    ///
488    /// The untyped [`send`](Self::send) method remains available at the CDP
489    /// boundary, while session code can use this helper for stable response
490    /// shapes without repeatedly indexing raw `serde_json::Value` objects.
491    pub async fn send_typed<R: DeserializeOwned>(
492        &self,
493        method: &str,
494        params: Option<Value>,
495    ) -> Result<R, CdpError> {
496        let value = self.send(method, params).await?;
497        serde_json::from_value(value).map_err(CdpError::decode)
498    }
499
500    /// Typed variant of [`send_browser`](Self::send_browser).
501    pub async fn send_browser_typed<R: DeserializeOwned>(
502        &self,
503        method: &str,
504        params: Option<Value>,
505    ) -> Result<R, CdpError> {
506        let value = self.send_browser(method, params).await?;
507        serde_json::from_value(value).map_err(CdpError::decode)
508    }
509
510    /// Typed variant of [`send_to_session`](Self::send_to_session).
511    pub async fn send_to_session_typed<R: DeserializeOwned>(
512        &self,
513        session_id: &str,
514        method: &str,
515        params: Option<Value>,
516    ) -> Result<R, CdpError> {
517        let value = self.send_to_session(session_id, method, params).await?;
518        serde_json::from_value(value).map_err(CdpError::decode)
519    }
520
521    pub async fn send_browser(
522        &self,
523        method: &str,
524        params: Option<Value>,
525    ) -> Result<Value, CdpError> {
526        self.send_routed(method, params, None, self.timeout).await
527    }
528
529    pub async fn send_to_session(
530        &self,
531        session_id: &str,
532        method: &str,
533        params: Option<Value>,
534    ) -> Result<Value, CdpError> {
535        self.send_routed(method, params, Some(session_id.to_string()), self.timeout)
536            .await
537    }
538
539    /// Send one operation-scoped command without changing the connection's
540    /// default response deadline or active route.
541    pub async fn send_with_timeout(
542        &self,
543        method: &str,
544        params: Option<Value>,
545        timeout: Duration,
546    ) -> Result<Value, CdpError> {
547        let session_id = self.current_route().session_id;
548        self.send_routed(method, params, session_id, timeout).await
549    }
550
551    async fn send_routed(
552        &self,
553        method: &str,
554        params: Option<Value>,
555        session_id: Option<String>,
556        timeout: Duration,
557    ) -> Result<Value, CdpError> {
558        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
559        let request = CdpRequest {
560            id,
561            method: method.to_string(),
562            params,
563            session_id,
564        };
565        let json = serde_json::to_string(&request)
566            .map_err(|error| CdpError::transport(format!("failed to encode request: {error}")))?;
567        let (response_tx, response_rx) = oneshot::channel();
568
569        self.tx
570            .send(Command::Request {
571                id,
572                json,
573                response: response_tx,
574            })
575            .map_err(|_| CdpError::transport("CDP connection task is unavailable"))?;
576        let mut pending_guard = PendingRequestGuard {
577            tx: self.tx.clone(),
578            id,
579            armed: true,
580        };
581        let started = std::time::Instant::now();
582
583        let result = match tokio::time::timeout(timeout, response_rx).await {
584            Ok(Ok(result)) => {
585                pending_guard.disarm();
586                result
587            }
588            Ok(Err(_)) => {
589                pending_guard.disarm();
590                Err(CdpError::transport("CDP response channel closed"))
591            }
592            Err(_) => Err(CdpError::response_timeout(timeout)),
593        };
594        let elapsed_nanos = started.elapsed().as_nanos().min(u64::MAX as u128) as u64;
595        self.cdp_wait_nanos
596            .fetch_add(elapsed_nanos, Ordering::Relaxed);
597        if let Ok(scope) = CDP_WAIT_SCOPE.try_with(Arc::clone) {
598            scope.fetch_add(elapsed_nanos, Ordering::Relaxed);
599        }
600        result
601    }
602
603    pub fn set_active_session(&self, session_id: Option<String>) {
604        let mut route = self
605            .active_route
606            .lock()
607            .unwrap_or_else(|poison| poison.into_inner());
608        route.session_id = session_id;
609        route.context_id = None;
610        route.frame_id = None;
611    }
612
613    pub fn set_active_context(&self, context_id: Option<i64>) {
614        self.active_route
615            .lock()
616            .unwrap_or_else(|poison| poison.into_inner())
617            .context_id = context_id;
618    }
619
620    pub fn set_active_frame_context(&self, frame_id: Option<String>, context_id: Option<i64>) {
621        let mut route = self
622            .active_route
623            .lock()
624            .unwrap_or_else(|poison| poison.into_inner());
625        route.frame_id = frame_id;
626        route.context_id = context_id;
627    }
628
629    pub fn set_active_route(
630        &self,
631        session_id: Option<String>,
632        frame_id: Option<String>,
633        context_id: Option<i64>,
634    ) {
635        let mut route = self
636            .active_route
637            .lock()
638            .unwrap_or_else(|poison| poison.into_inner());
639        route.session_id = session_id;
640        route.frame_id = frame_id;
641        route.context_id = context_id;
642    }
643
644    pub fn set_active_target_route(
645        &self,
646        target_id: Option<String>,
647        session_id: Option<String>,
648        frame_id: Option<String>,
649        context_id: Option<i64>,
650    ) {
651        *self
652            .active_route
653            .lock()
654            .unwrap_or_else(|poison| poison.into_inner()) = CdpRoute {
655            target_id,
656            session_id,
657            frame_id,
658            context_id,
659        };
660    }
661
662    pub fn operation_identity(&self) -> Option<(String, String)> {
663        let route = self.current_route();
664        Some((route.target_id?, route.frame_id?))
665    }
666
667    pub fn set_active_frame(&self, frame_id: Option<String>) {
668        self.active_route
669            .lock()
670            .unwrap_or_else(|poison| poison.into_inner())
671            .frame_id = frame_id;
672    }
673
674    pub fn active_frame(&self) -> Option<String> {
675        self.current_route().frame_id
676    }
677
678    fn current_route(&self) -> CdpRoute {
679        OPERATION_ROUTE.try_with(Clone::clone).unwrap_or_else(|_| {
680            self.active_route
681                .lock()
682                .unwrap_or_else(|poison| poison.into_inner())
683                .clone()
684        })
685    }
686
687    pub async fn with_current_route<F: std::future::Future>(&self, future: F) -> F::Output {
688        if OPERATION_ROUTE.try_with(|_| ()).is_ok() {
689            future.await
690        } else {
691            OPERATION_ROUTE.scope(self.current_route(), future).await
692        }
693    }
694
695    pub async fn with_current_target_route<F: std::future::Future>(&self, future: F) -> F::Output {
696        let mut route = self.current_route();
697        route.context_id = None;
698        OPERATION_ROUTE.scope(route, future).await
699    }
700
701    /// Return the selected child frame viewport origin in target coordinates.
702    pub async fn frame_viewport_offset(&self, frame_id: &str) -> Result<(f64, f64), CdpError> {
703        let owner = self
704            .send(
705                "DOM.getFrameOwner",
706                Some(serde_json::json!({"frameId": frame_id})),
707            )
708            .await?;
709        let backend_node_id = owner["backendNodeId"]
710            .as_i64()
711            .ok_or_else(|| CdpError::transport("frame owner contained no backend node ID"))?;
712        let model = self
713            .send(
714                "DOM.getBoxModel",
715                Some(serde_json::json!({"backendNodeId": backend_node_id})),
716            )
717            .await?;
718        let content = model["model"]["content"]
719            .as_array()
720            .filter(|quad| quad.len() >= 2)
721            .ok_or_else(|| CdpError::transport("frame owner contained no content quad"))?;
722        let x = content[0]
723            .as_f64()
724            .ok_or_else(|| CdpError::transport("frame owner x was not numeric"))?;
725        let y = content[1]
726            .as_f64()
727            .ok_or_else(|| CdpError::transport("frame owner y was not numeric"))?;
728        Ok((x, y))
729    }
730
731    /// Navigate to a URL.
732    pub async fn navigate(&self, url: &str) -> Result<Value, CdpError> {
733        self.send("Page.navigate", Some(serde_json::json!({ "url": url })))
734            .await
735    }
736
737    /// Take a screenshot and return its base64-encoded image data.
738    pub async fn screenshot(&self, format: &str) -> Result<String, CdpError> {
739        self.screenshot_with_params(serde_json::json!({
740            "format": format,
741            "optimizeForSpeed": true
742        }))
743        .await
744    }
745
746    pub async fn screenshot_with_params(&self, params: Value) -> Result<String, CdpError> {
747        let mut result = self.send("Page.captureScreenshot", Some(params)).await?;
748        match result.get_mut("data").map(Value::take) {
749            Some(Value::String(data)) => Ok(data),
750            _ => Err(CdpError::transport(
751                "CDP screenshot response contained no data",
752            )),
753        }
754    }
755
756    pub async fn get_layout_metrics(&self) -> Result<Value, CdpError> {
757        self.send("Page.getLayoutMetrics", None).await
758    }
759
760    /// Get the accessibility tree.
761    pub async fn get_accessibility_tree(&self) -> Result<Value, CdpError> {
762        let frame_id = self.current_route().frame_id;
763        self.send(
764            "Accessibility.getFullAXTree",
765            frame_id.map(|frame_id| serde_json::json!({"frameId": frame_id})),
766        )
767        .await
768    }
769
770    /// Get a flattened document tree including shadow DOM content.
771    /// Uses `pierce: true` to include open shadow roots up to the given depth.
772    pub async fn get_flattened_document(&self, depth: i64) -> Result<Value, CdpError> {
773        self.send(
774            "DOM.getFlattenedDocument",
775            Some(serde_json::json!({ "depth": depth, "pierce": true })),
776        )
777        .await
778    }
779
780    /// Get the full document tree for an explicit deep-DOM inspection.
781    pub async fn get_deep_document(&self) -> Result<Value, CdpError> {
782        self.send("DOM.getDocument", Some(serde_json::json!({ "depth": -1 })))
783            .await
784    }
785
786    /// Get only the document root for operations that do not need descendants.
787    pub async fn get_document_root(&self) -> Result<Value, CdpError> {
788        self.send("DOM.getDocument", Some(serde_json::json!({ "depth": 0 })))
789            .await
790    }
791
792    /// Compatibility alias for the explicit deep-DOM request.
793    pub async fn get_document(&self) -> Result<Value, CdpError> {
794        self.get_deep_document().await
795    }
796
797    /// Query a CSS selector and return the matching node.
798    pub async fn query_selector(&self, selector: &str) -> Result<Value, CdpError> {
799        let document = self.get_document_root().await?;
800        let root_id = document["root"]["nodeId"]
801            .as_i64()
802            .ok_or_else(|| CdpError::transport("DOM document response contained no root nodeId"))?;
803        self.send(
804            "DOM.querySelector",
805            Some(serde_json::json!({ "nodeId": root_id, "selector": selector })),
806        )
807        .await
808    }
809
810    /// Resolve a DOM/backend node to a reusable remote object.
811    pub async fn resolve_node_object(
812        &self,
813        node_id: Option<i64>,
814        backend_node_id: Option<i64>,
815    ) -> Result<String, CdpError> {
816        let mut params = serde_json::Map::new();
817        if let Some(node_id) = node_id {
818            params.insert("nodeId".to_string(), Value::from(node_id));
819        }
820        if let Some(backend_node_id) = backend_node_id {
821            params.insert("backendNodeId".to_string(), Value::from(backend_node_id));
822        }
823        let resolved = self
824            .send("DOM.resolveNode", Some(Value::Object(params)))
825            .await?;
826        resolved["object"]["objectId"]
827            .as_str()
828            .map(str::to_string)
829            .ok_or_else(|| CdpError::transport("DOM.resolveNode returned no objectId"))
830    }
831
832    /// Translate one already-resolved frontend node into its immutable backend
833    /// identity without repeating the caller's locator query.
834    pub async fn backend_node_id_for_node(&self, node_id: i64) -> Result<i64, CdpError> {
835        let described = self
836            .send(
837                "DOM.describeNode",
838                Some(serde_json::json!({"nodeId": node_id, "depth": 0})),
839            )
840            .await?;
841        described["node"]["backendNodeId"]
842            .as_i64()
843            .filter(|id| *id > 0)
844            .ok_or_else(|| CdpError::transport("DOM.describeNode returned no backendNodeId"))
845    }
846
847    /// Invoke a function on a previously resolved remote object.
848    pub async fn call_on_object(
849        &self,
850        object_id: &str,
851        function_declaration: &str,
852    ) -> Result<Value, CdpError> {
853        self.send(
854            "Runtime.callFunctionOn",
855            Some(serde_json::json!({
856                "objectId": object_id,
857                "functionDeclaration": function_declaration,
858                "returnByValue": true,
859                "awaitPromise": true
860            })),
861        )
862        .await
863    }
864
865    pub async fn release_object(&self, object_id: &str) -> Result<Value, CdpError> {
866        self.send(
867            "Runtime.releaseObject",
868            Some(serde_json::json!({ "objectId": object_id })),
869        )
870        .await
871    }
872
873    pub async fn release_object_for_session(
874        &self,
875        session_id: &str,
876        object_id: &str,
877    ) -> Result<Value, CdpError> {
878        self.send_to_session(
879            session_id,
880            "Runtime.releaseObject",
881            Some(serde_json::json!({"objectId": object_id})),
882        )
883        .await
884    }
885
886    /// Resolve a page-produced, bounded remote element array into DOM node IDs.
887    ///
888    /// The expression must return an Array with at most `limit` elements and a
889    /// numeric `glassCount` property containing the total logical match count.
890    pub async fn bounded_element_query(
891        &self,
892        expression: &str,
893        limit: usize,
894    ) -> Result<(usize, Vec<i64>), CdpError> {
895        // DOM.requestNode only returns frontend node IDs after the document has
896        // been requested in this CDP session.
897        self.get_document_root().await?;
898        let context_id = self.current_route().context_id;
899        let mut params = serde_json::json!({
900            "expression": expression,
901            "returnByValue": false,
902            "awaitPromise": true
903        });
904        if let Some(context_id) = context_id {
905            params["contextId"] = Value::from(context_id);
906        }
907        let evaluated = self.send("Runtime.evaluate", Some(params)).await?;
908        if evaluated.get("exceptionDetails").is_some() {
909            return Err(CdpError::transport("element query evaluation failed"));
910        }
911        let array_id = evaluated["result"]["objectId"]
912            .as_str()
913            .ok_or_else(|| CdpError::transport("element query returned no remote array"))?;
914        let mut remote_objects = RemoteObjectBatchGuard::new(self.clone(), array_id.to_string());
915        let properties = self
916            .send(
917                "Runtime.getProperties",
918                Some(serde_json::json!({
919                    "objectId": array_id,
920                    "ownProperties": true
921                })),
922            )
923            .await?;
924        let mut count = 0;
925        let mut objects = Vec::with_capacity(limit);
926        for property in properties["result"].as_array().into_iter().flatten() {
927            if property["name"].as_str() == Some("glassCount") {
928                count = property["value"]["value"].as_u64().unwrap_or(0) as usize;
929                continue;
930            }
931            if property["name"]
932                .as_str()
933                .and_then(|name| name.parse::<usize>().ok())
934                .is_some_and(|index| index < limit)
935                && let Some(object_id) = property["value"]["objectId"].as_str()
936            {
937                remote_objects.object_ids.push(object_id.to_string());
938                objects.push(object_id.to_string());
939            }
940        }
941        let mut node_ids = Vec::with_capacity(objects.len());
942        for object_id in objects {
943            let requested = self
944                .send(
945                    "DOM.requestNode",
946                    Some(serde_json::json!({ "objectId": object_id })),
947                )
948                .await;
949            let requested = requested?;
950            if let Some(node_id) = requested["nodeId"].as_i64().filter(|id| *id != 0) {
951                node_ids.push(node_id);
952            }
953        }
954        remote_objects.cleanup().await?;
955        Ok((count, node_ids))
956    }
957
958    /// Get the bounding box of a DOM node.
959    pub async fn get_box_model(&self, node_id: i64) -> Result<Value, CdpError> {
960        self.get_box_model_inner(Some(node_id), None).await
961    }
962
963    /// Get the bounding box of a backend DOM node from an accessibility tree.
964    pub async fn get_box_model_for_backend(&self, backend_node_id: i64) -> Result<Value, CdpError> {
965        self.get_box_model_inner(None, Some(backend_node_id)).await
966    }
967
968    /// Ask Chrome to scroll a DOM node into view only when it is necessary.
969    ///
970    /// The browser owns the visibility decision, avoiding a separate layout
971    /// probe and avoiding a scroll when the target is already actionable.
972    pub async fn scroll_into_view_if_needed(
973        &self,
974        node_id: Option<i64>,
975        backend_node_id: Option<i64>,
976    ) -> Result<Value, CdpError> {
977        let mut params = serde_json::Map::new();
978        if let Some(node_id) = node_id {
979            params.insert("nodeId".to_string(), Value::from(node_id));
980        }
981        if let Some(backend_node_id) = backend_node_id {
982            params.insert("backendNodeId".to_string(), Value::from(backend_node_id));
983        }
984        if params.is_empty() {
985            return Err(CdpError::transport(
986                "scrollIntoViewIfNeeded requires a nodeId or backendNodeId",
987            ));
988        }
989        self.send("DOM.scrollIntoViewIfNeeded", Some(Value::Object(params)))
990            .await
991    }
992
993    async fn get_box_model_inner(
994        &self,
995        node_id: Option<i64>,
996        backend_node_id: Option<i64>,
997    ) -> Result<Value, CdpError> {
998        let mut params = serde_json::Map::new();
999        if let Some(node_id) = node_id {
1000            params.insert("nodeId".to_string(), Value::from(node_id));
1001        }
1002        if let Some(backend_node_id) = backend_node_id {
1003            params.insert("backendNodeId".to_string(), Value::from(backend_node_id));
1004        }
1005        self.send("DOM.getBoxModel", Some(Value::Object(params)))
1006            .await
1007    }
1008
1009    /// Evaluate JavaScript in the page.
1010    pub async fn evaluate(&self, expression: &str) -> Result<Value, CdpError> {
1011        let context_id = self.current_route().context_id;
1012        self.evaluate_in_context(expression, context_id).await
1013    }
1014
1015    pub async fn evaluate_in_context(
1016        &self,
1017        expression: &str,
1018        context_id: Option<i64>,
1019    ) -> Result<Value, CdpError> {
1020        let mut params = serde_json::json!({
1021            "expression": expression,
1022            "returnByValue": true,
1023            "awaitPromise": true
1024        });
1025        if let Some(context_id) = context_id {
1026            params["contextId"] = Value::from(context_id);
1027        }
1028        self.send("Runtime.evaluate", Some(params)).await
1029    }
1030
1031    /// Insert text into the currently focused element.
1032    pub async fn insert_text(&self, text: &str) -> Result<Value, CdpError> {
1033        self.send(
1034            "Input.insertText",
1035            Some(serde_json::json!({ "text": text })),
1036        )
1037        .await
1038    }
1039
1040    /// Dispatch a mouse event via CDP Input.
1041    pub async fn dispatch_mouse_event(
1042        &self,
1043        event_type: &str,
1044        x: f64,
1045        y: f64,
1046        button: Option<&str>,
1047        click_count: Option<u32>,
1048    ) -> Result<Value, CdpError> {
1049        let mut params = serde_json::json!({
1050            "type": event_type,
1051            "x": x,
1052            "y": y,
1053        });
1054        if let Some(button) = button {
1055            params["button"] = Value::from(button);
1056        }
1057        if let Some(click_count) = click_count {
1058            params["clickCount"] = Value::from(click_count);
1059        }
1060        self.send("Input.dispatchMouseEvent", Some(params)).await
1061    }
1062
1063    /// Dispatch one mouse event with a caller-owned response window.
1064    ///
1065    /// This does not alter the connection default used by ordinary input.
1066    pub async fn dispatch_mouse_event_with_timeout(
1067        &self,
1068        event_type: &str,
1069        x: f64,
1070        y: f64,
1071        button: Option<&str>,
1072        click_count: Option<u32>,
1073        timeout: Duration,
1074    ) -> Result<Value, CdpError> {
1075        let mut params = serde_json::json!({"type": event_type, "x": x, "y": y});
1076        if let Some(button) = button {
1077            params["button"] = Value::from(button);
1078        }
1079        if let Some(click_count) = click_count {
1080            params["clickCount"] = Value::from(click_count);
1081        }
1082        self.send_with_timeout("Input.dispatchMouseEvent", Some(params), timeout)
1083            .await
1084    }
1085
1086    /// Dispatch a keyboard event via CDP Input.
1087    pub async fn dispatch_key_event(
1088        &self,
1089        event_type: &str,
1090        key: &str,
1091        code: &str,
1092    ) -> Result<Value, CdpError> {
1093        self.send(
1094            "Input.dispatchKeyEvent",
1095            Some(serde_json::json!({
1096                "type": event_type,
1097                "key": key,
1098                "code": code,
1099                "text": if event_type == "keyDown" { key } else { "" }
1100            })),
1101        )
1102        .await
1103    }
1104
1105    pub async fn dispatch_key_event_with_modifiers(
1106        &self,
1107        event_type: &str,
1108        key: &str,
1109        code: &str,
1110        text: &str,
1111        modifiers: i64,
1112    ) -> Result<Value, CdpError> {
1113        let virtual_key_code = match key {
1114            "Backspace" => 8,
1115            "Tab" => 9,
1116            "Enter" => 13,
1117            "Escape" => 27,
1118            "Delete" => 46,
1119            _ if key.len() == 1 => key.as_bytes()[0].to_ascii_uppercase() as i64,
1120            _ => 0,
1121        };
1122        self.send(
1123            "Input.dispatchKeyEvent",
1124            Some(serde_json::json!({
1125                "type": event_type,
1126                "key": key,
1127                "code": code,
1128                "text": text,
1129                "modifiers": modifiers,
1130                "windowsVirtualKeyCode": virtual_key_code,
1131                "nativeVirtualKeyCode": virtual_key_code
1132            })),
1133        )
1134        .await
1135    }
1136
1137    /// Invoke Blink's platform-independent editing command on the focused node.
1138    pub async fn dispatch_select_all(&self) -> Result<Value, CdpError> {
1139        self.send(
1140            "Input.dispatchKeyEvent",
1141            Some(serde_json::json!({
1142                "type": "rawKeyDown",
1143                "key": "a",
1144                "code": "KeyA",
1145                "commands": ["selectAll"]
1146            })),
1147        )
1148        .await
1149    }
1150
1151    pub async fn set_file_input_files(
1152        &self,
1153        node_id: Option<i64>,
1154        backend_node_id: Option<i64>,
1155        files: &[String],
1156    ) -> Result<Value, CdpError> {
1157        let mut params = serde_json::json!({"files": files});
1158        if let Some(node_id) = node_id {
1159            params["nodeId"] = Value::from(node_id);
1160        }
1161        if let Some(backend_node_id) = backend_node_id {
1162            params["backendNodeId"] = Value::from(backend_node_id);
1163        }
1164        self.send("DOM.setFileInputFiles", Some(params)).await
1165    }
1166
1167    /// Scroll the current page by a delta in CSS pixels.
1168    pub async fn scroll_by(&self, dx: f64, dy: f64) -> Result<Value, CdpError> {
1169        let expression = format!(
1170            "window.scrollBy({:.4}, {:.4}); window.scrollX + ',' + window.scrollY",
1171            dx, dy
1172        );
1173        self.evaluate(&expression).await
1174    }
1175
1176    pub async fn get_cookies(&self) -> Result<Value, CdpError> {
1177        self.send("Network.getCookies", None).await
1178    }
1179
1180    pub async fn set_cookies(&self, cookies: Value) -> Result<Value, CdpError> {
1181        self.send(
1182            "Network.setCookies",
1183            Some(serde_json::json!({ "cookies": cookies })),
1184        )
1185        .await
1186    }
1187
1188    pub async fn clear_browser_cookies(&self) -> Result<(), CdpError> {
1189        self.send("Network.clearBrowserCookies", None).await?;
1190        Ok(())
1191    }
1192    pub async fn enable_page(&self) -> Result<(), CdpError> {
1193        self.send("Page.enable", None).await?;
1194        Ok(())
1195    }
1196
1197    /// Enable the only event domains required to wait for navigation and
1198    /// invalidate compact observations.
1199    pub async fn enable_observation_events(&self) -> Result<(), CdpError> {
1200        self.enable_page().await?;
1201        self.enable_dom().await?;
1202        Ok(())
1203    }
1204
1205    pub async fn enable_observation_events_for(&self, session_id: &str) -> Result<(), CdpError> {
1206        self.send_to_session(session_id, "Page.enable", None)
1207            .await?;
1208        self.send_to_session(session_id, "DOM.enable", None).await?;
1209        Ok(())
1210    }
1211
1212    pub async fn enable_runtime(&self) -> Result<(), CdpError> {
1213        self.send("Runtime.enable", None).await?;
1214        Ok(())
1215    }
1216
1217    pub async fn disable_runtime(&self) -> Result<(), CdpError> {
1218        self.send("Runtime.disable", None).await?;
1219        Ok(())
1220    }
1221
1222    pub async fn enable_log(&self) -> Result<(), CdpError> {
1223        self.send("Log.enable", None).await?;
1224        Ok(())
1225    }
1226
1227    pub async fn disable_log(&self) -> Result<(), CdpError> {
1228        self.send("Log.disable", None).await?;
1229        Ok(())
1230    }
1231
1232    pub async fn enable_network(&self) -> Result<(), CdpError> {
1233        self.send("Network.enable", None).await?;
1234        Ok(())
1235    }
1236
1237    pub async fn disable_network(&self) -> Result<(), CdpError> {
1238        self.send("Network.disable", None).await?;
1239        Ok(())
1240    }
1241
1242    pub async fn handle_javascript_dialog(&self, accept: bool) -> Result<Value, CdpError> {
1243        self.send(
1244            "Page.handleJavaScriptDialog",
1245            Some(serde_json::json!({"accept": accept})),
1246        )
1247        .await
1248    }
1249
1250    pub async fn set_download_behavior(
1251        &self,
1252        behavior: &str,
1253        download_path: Option<&Path>,
1254        events_enabled: bool,
1255    ) -> Result<Value, CdpError> {
1256        let mut params = serde_json::json!({
1257            "behavior": behavior,
1258            "eventsEnabled": events_enabled
1259        });
1260        if let Some(path) = download_path {
1261            params["downloadPath"] = Value::from(path.to_string_lossy().into_owned());
1262        }
1263        self.send_browser("Browser.setDownloadBehavior", Some(params))
1264            .await
1265    }
1266
1267    pub async fn enable_dom(&self) -> Result<(), CdpError> {
1268        self.send("DOM.enable", None).await?;
1269        Ok(())
1270    }
1271
1272    pub async fn enable_accessibility(&self) -> Result<(), CdpError> {
1273        self.send("Accessibility.enable", None).await?;
1274        Ok(())
1275    }
1276
1277    /// Ask an owned Chrome browser to close itself before process-level
1278    /// shutdown. This gives profile-backed state a chance to flush cleanly.
1279    pub async fn close_browser(&self) -> Result<(), CdpError> {
1280        self.send_browser("Browser.close", None).await?;
1281        Ok(())
1282    }
1283
1284    /// Override device metrics for viewport emulation.
1285    pub async fn set_device_metrics_override(
1286        &self,
1287        width: i64,
1288        height: i64,
1289        device_scale_factor: f64,
1290        mobile: bool,
1291    ) -> Result<Value, CdpError> {
1292        self.send(
1293            "Emulation.setDeviceMetricsOverride",
1294            Some(serde_json::json!({
1295                "width": width,
1296                "height": height,
1297                "deviceScaleFactor": device_scale_factor,
1298                "mobile": mobile,
1299            })),
1300        )
1301        .await
1302    }
1303
1304    /// Clear device metrics override.
1305    pub async fn clear_device_metrics_override(&self) -> Result<Value, CdpError> {
1306        self.send("Emulation.clearDeviceMetricsOverride", None)
1307            .await
1308    }
1309
1310    /// Ask the connection task to close its WebSocket.
1311    pub async fn close(&self) {
1312        let _ = self.tx.send(Command::Close);
1313    }
1314}
1315
1316struct ScreencastDispatch<'a> {
1317    sink: &'a std::sync::Mutex<Option<ScreencastSink>>,
1318    received: &'a AtomicU64,
1319    dropped: &'a AtomicU64,
1320    command_tx: &'a mpsc::UnboundedSender<Command>,
1321    next_id: &'a AtomicU64,
1322}
1323
1324fn handle_incoming_message(
1325    pending: &mut HashMap<u64, oneshot::Sender<Result<Value, CdpError>>>,
1326    events: &broadcast::Sender<CdpEvent>,
1327    payload_events: &broadcast::Sender<CdpEventWithParams>,
1328    screencast: ScreencastDispatch<'_>,
1329    text: &str,
1330) {
1331    let message: IncomingMessage = match serde_json::from_str(text) {
1332        Ok(message) => message,
1333        Err(error) => {
1334            warn!(%error, "ignoring malformed CDP message");
1335            return;
1336        }
1337    };
1338
1339    if let Some(id) = message.id {
1340        if let Some(response) = pending.remove(&id) {
1341            let result = match message.error {
1342                Some(error) => Err(error),
1343                None => Ok(message.result.unwrap_or(Value::Null)),
1344            };
1345            let _ = response.send(result);
1346        } else {
1347            debug!(id, "received CDP response with no pending request");
1348        }
1349        return;
1350    }
1351
1352    if let Some(method) = message.method {
1353        if method == "Page.screencastFrame" {
1354            match serde_json::from_str::<IncomingEventParams>(text) {
1355                Ok(mut payload) => {
1356                    let frame_session_id = payload.params["sessionId"].as_u64();
1357                    if let Some(frame_session_id) = frame_session_id {
1358                        let id = screencast.next_id.fetch_add(1, Ordering::Relaxed);
1359                        let mut ack = serde_json::json!({
1360                            "id": id,
1361                            "method": "Page.screencastFrameAck",
1362                            "params": {"sessionId": frame_session_id}
1363                        });
1364                        if let Some(session_id) = message.session_id.as_deref() {
1365                            ack["sessionId"] = Value::from(session_id);
1366                        }
1367                        let _ = screencast.command_tx.send(Command::FireAndForget {
1368                            json: ack.to_string(),
1369                        });
1370                    }
1371                    let data = payload.params["data"].take();
1372                    let metadata = payload.params["metadata"].take();
1373                    let frame = match data {
1374                        Value::String(data) => Some(CdpScreencastFrame {
1375                            data,
1376                            metadata,
1377                            session_id: message.session_id,
1378                        }),
1379                        _ => None,
1380                    };
1381                    if let Some(frame) = frame {
1382                        let sink = screencast.sink.lock().expect("screencast sink poisoned");
1383                        if let Some(sink) = sink.as_ref()
1384                            && sink.session_id == frame.session_id
1385                        {
1386                            screencast.received.fetch_add(1, Ordering::Relaxed);
1387                            if frame.data.len() > 32 * 1024 * 1024
1388                                || sink.sender.try_send(frame).is_err()
1389                            {
1390                                screencast.dropped.fetch_add(1, Ordering::Relaxed);
1391                            }
1392                        }
1393                    }
1394                }
1395                Err(error) => warn!(%error, "ignoring malformed screencast payload"),
1396            }
1397            let _ = events.send(CdpEvent { method });
1398            return;
1399        }
1400        let _ = events.send(CdpEvent {
1401            method: method.clone(),
1402        });
1403        if payload_events.receiver_count() > 0 {
1404            match serde_json::from_str::<IncomingEventParams>(text) {
1405                Ok(payload) => {
1406                    let _ = payload_events.send(CdpEventWithParams {
1407                        method,
1408                        params: payload.params,
1409                        session_id: message.session_id,
1410                    });
1411                }
1412                Err(error) => warn!(%error, "ignoring malformed CDP event payload"),
1413            }
1414        }
1415    }
1416}
1417
1418/// Exercise the production CDP envelope and event payload decoders.
1419#[cfg(feature = "fuzzing")]
1420#[doc(hidden)]
1421pub fn fuzz_incoming_message(text: &str) {
1422    if let Ok(message) = serde_json::from_str::<IncomingMessage>(text)
1423        && message.method.is_some()
1424    {
1425        let _ = serde_json::from_str::<IncomingEventParams>(text);
1426    }
1427}
1428
1429#[cfg(test)]
1430mod tests {
1431    use super::*;
1432    use futures_util::{SinkExt, StreamExt};
1433    use tokio::net::TcpListener;
1434    use tokio_tungstenite::accept_async;
1435
1436    #[tokio::test]
1437    async fn routes_concurrent_responses_by_id_and_delivers_events() {
1438        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1439        let address = listener.local_addr().unwrap();
1440
1441        let server = tokio::spawn(async move {
1442            let (stream, _) = listener.accept().await.unwrap();
1443            let websocket = accept_async(stream).await.unwrap();
1444            let (mut write, mut read) = websocket.split();
1445
1446            let first = read.next().await.unwrap().unwrap();
1447            let second = read.next().await.unwrap().unwrap();
1448            let first: Value = match first {
1449                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1450                _ => panic!("expected text frame"),
1451            };
1452            let second: Value = match second {
1453                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1454                _ => panic!("expected text frame"),
1455            };
1456
1457            write
1458                .send(Message::Text(
1459                    serde_json::json!({
1460                        "method": "Page.loadEventFired",
1461                        "params": {"frameId": "main"}
1462                    })
1463                    .to_string()
1464                    .into(),
1465                ))
1466                .await
1467                .unwrap();
1468
1469            for request in [second, first] {
1470                write
1471                    .send(Message::Text(
1472                        serde_json::json!({
1473                            "id": request["id"],
1474                            "result": {"method": request["method"]}
1475                        })
1476                        .to_string()
1477                        .into(),
1478                    ))
1479                    .await
1480                    .unwrap();
1481            }
1482        });
1483
1484        let client =
1485            CdpClient::connect_with_timeout(&format!("ws://{address}"), Duration::from_secs(2))
1486                .await
1487                .unwrap();
1488        let mut events = client.subscribe_events();
1489
1490        let (first, second) =
1491            tokio::join!(client.send("first", None), client.send("second", None),);
1492        assert_eq!(first.unwrap()["method"], "first");
1493        assert_eq!(second.unwrap()["method"], "second");
1494        assert_eq!(events.recv().await.unwrap().method, "Page.loadEventFired");
1495
1496        client.close().await;
1497        server.await.unwrap();
1498    }
1499
1500    #[tokio::test]
1501    async fn operation_route_is_immutable_across_selection_changes() {
1502        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1503        let address = listener.local_addr().unwrap();
1504        let server = tokio::spawn(async move {
1505            let (stream, _) = listener.accept().await.unwrap();
1506            let mut websocket = accept_async(stream).await.unwrap();
1507            for expected_session in ["old", "old", "new"] {
1508                let request = websocket.next().await.unwrap().unwrap();
1509                let request: Value = match request {
1510                    Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1511                    _ => panic!("expected text frame"),
1512                };
1513                assert_eq!(request["sessionId"], expected_session);
1514                websocket
1515                    .send(Message::Text(
1516                        serde_json::json!({"id": request["id"], "result": {}})
1517                            .to_string()
1518                            .into(),
1519                    ))
1520                    .await
1521                    .unwrap();
1522            }
1523        });
1524        let client = CdpClient::connect(&format!("ws://{address}"))
1525            .await
1526            .unwrap();
1527        client.set_active_target_route(
1528            Some("old-target".to_string()),
1529            Some("old".to_string()),
1530            Some("old-frame".to_string()),
1531            None,
1532        );
1533        client
1534            .with_current_route(async {
1535                client.send("first", None).await.unwrap();
1536                client.set_active_target_route(
1537                    Some("new-target".to_string()),
1538                    Some("new".to_string()),
1539                    Some("new-frame".to_string()),
1540                    None,
1541                );
1542                assert_eq!(
1543                    client.operation_identity(),
1544                    Some(("old-target".to_string(), "old-frame".to_string()))
1545                );
1546                client.send("second", None).await.unwrap();
1547            })
1548            .await;
1549        client.send("third", None).await.unwrap();
1550        client.close().await;
1551        server.await.unwrap();
1552    }
1553
1554    #[tokio::test]
1555    async fn delivers_event_payloads_only_to_opt_in_subscribers() {
1556        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1557        let address = listener.local_addr().unwrap();
1558        let server = tokio::spawn(async move {
1559            let (stream, _) = listener.accept().await.unwrap();
1560            let mut websocket = accept_async(stream).await.unwrap();
1561            let request = websocket.next().await.unwrap().unwrap();
1562            let request: Value = match request {
1563                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1564                _ => panic!("expected text frame"),
1565            };
1566
1567            for index in 0..5 {
1568                let route = if index == 0 { "foreign" } else { "wanted" };
1569                websocket
1570                    .send(Message::Text(
1571                        serde_json::json!({
1572                            "method": "Page.screencastFrame",
1573                            "sessionId": route,
1574                            "params": {"sessionId": 9, "data": format!("frame-{index}")}
1575                        })
1576                        .to_string()
1577                        .into(),
1578                    ))
1579                    .await
1580                    .unwrap();
1581                let ack = websocket.next().await.unwrap().unwrap();
1582                let ack: Value = match ack {
1583                    Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1584                    _ => panic!("expected text frame"),
1585                };
1586                assert_eq!(ack["method"], "Page.screencastFrameAck");
1587                assert_eq!(ack["params"]["sessionId"], 9);
1588                assert_eq!(ack["sessionId"], route);
1589            }
1590            websocket
1591                .send(Message::Text(
1592                    serde_json::json!({"id": request["id"], "result": {}})
1593                        .to_string()
1594                        .into(),
1595                ))
1596                .await
1597                .unwrap();
1598        });
1599
1600        let client = CdpClient::connect(&format!("ws://{address}"))
1601            .await
1602            .unwrap();
1603        let mut methods = client.subscribe_events();
1604        let mut payloads = client.subscribe_events_with_params();
1605        let mut frames = client
1606            .open_screencast_channel(Some("wanted".to_string()))
1607            .unwrap();
1608        client.send("test.ready", None).await.unwrap();
1609
1610        assert_eq!(methods.recv().await.unwrap().method, "Page.screencastFrame");
1611        assert_eq!(frames.recv().await.unwrap().data, "frame-1");
1612        assert_eq!(frames.recv().await.unwrap().data, "frame-2");
1613        assert!(payloads.try_recv().is_err());
1614        assert_eq!(client.screencast_stats(), (4, 2));
1615
1616        client.close().await;
1617        server.await.unwrap();
1618    }
1619
1620    #[tokio::test]
1621    async fn requests_fast_screenshot_encoding_and_moves_the_payload() {
1622        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1623        let address = listener.local_addr().unwrap();
1624        let server = tokio::spawn(async move {
1625            let (stream, _) = listener.accept().await.unwrap();
1626            let mut websocket = accept_async(stream).await.unwrap();
1627            let request = websocket.next().await.unwrap().unwrap();
1628            let request: Value = match request {
1629                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1630                _ => panic!("expected text frame"),
1631            };
1632            assert_eq!(request["method"], "Page.captureScreenshot");
1633            assert_eq!(request["params"]["format"], "png");
1634            assert_eq!(request["params"]["optimizeForSpeed"], true);
1635            websocket
1636                .send(Message::Text(
1637                    serde_json::json!({
1638                        "id": request["id"],
1639                        "result": {"data": "cG5n"}
1640                    })
1641                    .to_string()
1642                    .into(),
1643                ))
1644                .await
1645                .unwrap();
1646        });
1647
1648        let client = CdpClient::connect(&format!("ws://{address}"))
1649            .await
1650            .unwrap();
1651        assert_eq!(client.screenshot("png").await.unwrap(), "cG5n");
1652        client.close().await;
1653        server.await.unwrap();
1654    }
1655
1656    #[tokio::test]
1657    async fn selector_lookup_fetches_only_the_document_root() {
1658        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1659        let address = listener.local_addr().unwrap();
1660        let server = tokio::spawn(async move {
1661            let (stream, _) = listener.accept().await.unwrap();
1662            let mut websocket = accept_async(stream).await.unwrap();
1663
1664            let root_request = websocket.next().await.unwrap().unwrap();
1665            let root_request: Value = match root_request {
1666                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1667                _ => panic!("expected text frame"),
1668            };
1669            assert_eq!(root_request["method"], "DOM.getDocument");
1670            assert_eq!(root_request["params"], serde_json::json!({ "depth": 0 }));
1671            websocket
1672                .send(Message::Text(
1673                    serde_json::json!({
1674                        "id": root_request["id"],
1675                        "result": {"root": {"nodeId": 42}}
1676                    })
1677                    .to_string()
1678                    .into(),
1679                ))
1680                .await
1681                .unwrap();
1682
1683            let selector_request = websocket.next().await.unwrap().unwrap();
1684            let selector_request: Value = match selector_request {
1685                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1686                _ => panic!("expected text frame"),
1687            };
1688            assert_eq!(selector_request["method"], "DOM.querySelector");
1689            assert_eq!(
1690                selector_request["params"],
1691                serde_json::json!({ "nodeId": 42, "selector": "#save" })
1692            );
1693            websocket
1694                .send(Message::Text(
1695                    serde_json::json!({
1696                        "id": selector_request["id"],
1697                        "result": {"nodeId": 7}
1698                    })
1699                    .to_string()
1700                    .into(),
1701                ))
1702                .await
1703                .unwrap();
1704        });
1705
1706        let client = CdpClient::connect(&format!("ws://{address}"))
1707            .await
1708            .unwrap();
1709        assert_eq!(client.query_selector("#save").await.unwrap()["nodeId"], 7);
1710        client.close().await;
1711        server.await.unwrap();
1712    }
1713
1714    #[tokio::test]
1715    async fn backend_identity_describes_the_existing_frontend_node_without_a_second_query() {
1716        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1717        let address = listener.local_addr().unwrap();
1718        let server = tokio::spawn(async move {
1719            let (stream, _) = listener.accept().await.unwrap();
1720            let mut websocket = accept_async(stream).await.unwrap();
1721            let request = websocket.next().await.unwrap().unwrap();
1722            let request: Value = match request {
1723                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1724                _ => panic!("expected text frame"),
1725            };
1726            assert_eq!(request["method"], "DOM.describeNode");
1727            assert_eq!(
1728                request["params"],
1729                serde_json::json!({"nodeId": 17, "depth": 0})
1730            );
1731            websocket
1732                .send(Message::Text(
1733                    serde_json::json!({
1734                        "id": request["id"],
1735                        "result": {"node": {"backendNodeId": 91}}
1736                    })
1737                    .to_string()
1738                    .into(),
1739                ))
1740                .await
1741                .unwrap();
1742            assert!(
1743                tokio::time::timeout(Duration::from_millis(25), websocket.next())
1744                    .await
1745                    .is_err(),
1746                "backend translation must not repeat the selector query"
1747            );
1748        });
1749        let client = CdpClient::connect(&format!("ws://{address}"))
1750            .await
1751            .unwrap();
1752        assert_eq!(client.backend_node_id_for_node(17).await.unwrap(), 91);
1753        server.await.unwrap();
1754        client.close().await;
1755    }
1756
1757    #[tokio::test]
1758    async fn backend_identity_rejects_a_describe_response_without_backend_id() {
1759        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1760        let address = listener.local_addr().unwrap();
1761        let server = tokio::spawn(async move {
1762            let (stream, _) = listener.accept().await.unwrap();
1763            let mut websocket = accept_async(stream).await.unwrap();
1764            let request = websocket.next().await.unwrap().unwrap();
1765            let request: Value = match request {
1766                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1767                _ => panic!("expected text frame"),
1768            };
1769            websocket
1770                .send(Message::Text(
1771                    serde_json::json!({"id": request["id"], "result": {"node": {}}})
1772                        .to_string()
1773                        .into(),
1774                ))
1775                .await
1776                .unwrap();
1777        });
1778        let client = CdpClient::connect(&format!("ws://{address}"))
1779            .await
1780            .unwrap();
1781        let error = client.backend_node_id_for_node(17).await.unwrap_err();
1782        assert!(error.message.contains("no backendNodeId"));
1783        client.close().await;
1784        server.await.unwrap();
1785    }
1786
1787    #[tokio::test]
1788    async fn scroll_into_view_uses_the_backend_node_without_a_layout_probe() {
1789        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1790        let address = listener.local_addr().unwrap();
1791        let server = tokio::spawn(async move {
1792            let (stream, _) = listener.accept().await.unwrap();
1793            let mut websocket = accept_async(stream).await.unwrap();
1794            let request = websocket.next().await.unwrap().unwrap();
1795            let request: Value = match request {
1796                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1797                _ => panic!("expected text frame"),
1798            };
1799            assert_eq!(request["method"], "DOM.scrollIntoViewIfNeeded");
1800            assert_eq!(
1801                request["params"],
1802                serde_json::json!({ "backendNodeId": 42 })
1803            );
1804            websocket
1805                .send(Message::Text(
1806                    serde_json::json!({"id": request["id"], "result": {}})
1807                        .to_string()
1808                        .into(),
1809                ))
1810                .await
1811                .unwrap();
1812        });
1813
1814        let client = CdpClient::connect(&format!("ws://{address}"))
1815            .await
1816            .unwrap();
1817        client
1818            .scroll_into_view_if_needed(None, Some(42))
1819            .await
1820            .unwrap();
1821        client.close().await;
1822        server.await.unwrap();
1823    }
1824
1825    #[tokio::test]
1826    async fn observation_event_setup_enables_only_page_and_dom() {
1827        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1828        let address = listener.local_addr().unwrap();
1829        let server = tokio::spawn(async move {
1830            let (stream, _) = listener.accept().await.unwrap();
1831            let mut websocket = accept_async(stream).await.unwrap();
1832            let mut methods = Vec::new();
1833
1834            for _ in 0..2 {
1835                let request = websocket.next().await.unwrap().unwrap();
1836                let request: Value = match request {
1837                    Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1838                    _ => panic!("expected text frame"),
1839                };
1840                methods.push(request["method"].as_str().unwrap().to_string());
1841                websocket
1842                    .send(Message::Text(
1843                        serde_json::json!({"id": request["id"], "result": {}})
1844                            .to_string()
1845                            .into(),
1846                    ))
1847                    .await
1848                    .unwrap();
1849            }
1850
1851            assert_eq!(methods, ["Page.enable", "DOM.enable"]);
1852        });
1853
1854        let client = CdpClient::connect(&format!("ws://{address}"))
1855            .await
1856            .unwrap();
1857        client.enable_observation_events().await.unwrap();
1858        client.close().await;
1859        server.await.unwrap();
1860    }
1861
1862    #[tokio::test]
1863    async fn sends_browser_close_for_owned_session_shutdown() {
1864        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1865        let address = listener.local_addr().unwrap();
1866        let server = tokio::spawn(async move {
1867            let (stream, _) = listener.accept().await.unwrap();
1868            let mut websocket = accept_async(stream).await.unwrap();
1869            let request = websocket.next().await.unwrap().unwrap();
1870            let request: Value = match request {
1871                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1872                _ => panic!("expected text frame"),
1873            };
1874            assert_eq!(request["method"], "Browser.close");
1875            assert!(request.get("params").is_none());
1876            websocket
1877                .send(Message::Text(
1878                    serde_json::json!({"id": request["id"], "result": {}})
1879                        .to_string()
1880                        .into(),
1881                ))
1882                .await
1883                .unwrap();
1884        });
1885
1886        let client = CdpClient::connect(&format!("ws://{address}"))
1887            .await
1888            .unwrap();
1889        client.close_browser().await.unwrap();
1890        client.close().await;
1891        server.await.unwrap();
1892    }
1893
1894    #[tokio::test]
1895    async fn returns_a_timeout_when_the_server_does_not_respond() {
1896        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1897        let address = listener.local_addr().unwrap();
1898        let server = tokio::spawn(async move {
1899            let (stream, _) = listener.accept().await.unwrap();
1900            let (_write, mut read) = accept_async(stream).await.unwrap().split();
1901            let _ = read.next().await;
1902            tokio::time::sleep(Duration::from_millis(200)).await;
1903        });
1904
1905        let client =
1906            CdpClient::connect_with_timeout(&format!("ws://{address}"), Duration::from_millis(50))
1907                .await
1908                .unwrap();
1909        let error = client.send("never", None).await.unwrap_err();
1910        assert!(error.message.contains("timeout"));
1911        assert!(error.is_response_timeout());
1912        client.close().await;
1913        server.await.unwrap();
1914    }
1915
1916    #[tokio::test]
1917    async fn operation_timeout_is_short_and_does_not_change_the_connection_default() {
1918        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1919        let address = listener.local_addr().unwrap();
1920        let server = tokio::spawn(async move {
1921            let (stream, _) = listener.accept().await.unwrap();
1922            let mut websocket = accept_async(stream).await.unwrap();
1923            let first = websocket.next().await.unwrap().unwrap();
1924            let first: Value = match first {
1925                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1926                _ => panic!("expected text frame"),
1927            };
1928            assert_eq!(first["method"], "short");
1929            let second = websocket.next().await.unwrap().unwrap();
1930            let second: Value = match second {
1931                Message::Text(text) => serde_json::from_str(text.as_ref()).unwrap(),
1932                _ => panic!("expected text frame"),
1933            };
1934            assert_eq!(second["method"], "ordinary");
1935            websocket
1936                .send(Message::Text(
1937                    serde_json::json!({"id": second["id"], "result": {"ok": true}})
1938                        .to_string()
1939                        .into(),
1940                ))
1941                .await
1942                .unwrap();
1943        });
1944
1945        let client =
1946            CdpClient::connect_with_timeout(&format!("ws://{address}"), Duration::from_millis(500))
1947                .await
1948                .unwrap();
1949        let started = tokio::time::Instant::now();
1950        let error = client
1951            .send_with_timeout("short", None, Duration::from_millis(20))
1952            .await
1953            .unwrap_err();
1954        assert!(error.is_response_timeout());
1955        assert!(started.elapsed() < Duration::from_millis(200));
1956        assert_eq!(client.send("ordinary", None).await.unwrap()["ok"], true);
1957        client.close().await;
1958        server.await.unwrap();
1959    }
1960
1961    #[test]
1962    fn protocol_errors_are_not_typed_as_response_timeouts() {
1963        let error: CdpError = serde_json::from_value(serde_json::json!({
1964            "code": -32000,
1965            "message": "some protocol failure"
1966        }))
1967        .unwrap();
1968        assert!(!error.is_response_timeout());
1969    }
1970}
glass/browser/cdp.rs

glass/browser/
cdp.rs
