mirror of
https://github.com/lexmount/moli.git
synced 2026-10-30 08:07:38 +00:00
refactor(cdp): dispatch browser close through normal frontend routing
Send Browser.close through the ordinary command channel and Router session validation before handling it in the protocol actor. Route its response through the existing pending-command table and confirm writer flush before submitting InstanceCommand::Shutdown, without scheduling renderer work. Remove the socket shortcut, CdpFrontendSessionLookup, and shared Arc/Weak routing state. Inject one instance control handle into both server admission and owner lifecycle, keeping native shutdown on the same state. Cover response ID/session restoration and route consumption; retain failed flush, invalid/foreign session, pending-promise, and instance cleanup coverage.
This commit is contained in:
@@ -2,7 +2,6 @@ use anyhow::{Context, Result, bail};
|
||||
use moli_protocol_cdp::{CdpCommandParseError, ParsedCdpCommand};
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
|
||||
use crate::cdp_frontend_router::{CdpFrontendRouter, CdpFrontendSessionLookup};
|
||||
use crate::cdp_writer::CdpSocketSink;
|
||||
|
||||
const FRONTEND_COMMAND_QUEUE_CAPACITY: usize = 256;
|
||||
@@ -17,7 +16,6 @@ pub(crate) struct CdpFrontendEndpoint {
|
||||
control_tx: mpsc::UnboundedSender<CdpFrontendControlRequest>,
|
||||
command_tx: mpsc::Sender<CdpFrontendCommand>,
|
||||
shutdown_tx: tokio::sync::watch::Sender<bool>,
|
||||
session_lookup: CdpFrontendSessionLookup,
|
||||
}
|
||||
|
||||
pub(crate) enum CdpFrontendControlRequest {
|
||||
@@ -68,24 +66,20 @@ pub(crate) struct CdpCreatedTarget {
|
||||
pub(crate) tab_target_id: String,
|
||||
}
|
||||
|
||||
pub(crate) fn cdp_frontend_channel()
|
||||
-> (CdpFrontendEndpoint, CdpFrontendReceivers, CdpFrontendRouter) {
|
||||
pub(crate) fn cdp_frontend_channel() -> (CdpFrontendEndpoint, CdpFrontendReceivers) {
|
||||
let (control_tx, control_rx) = mpsc::unbounded_channel();
|
||||
let (command_tx, command_rx) = mpsc::channel(FRONTEND_COMMAND_QUEUE_CAPACITY);
|
||||
let (shutdown_tx, _) = tokio::sync::watch::channel(false);
|
||||
let router = CdpFrontendRouter::new();
|
||||
(
|
||||
CdpFrontendEndpoint {
|
||||
control_tx,
|
||||
command_tx,
|
||||
shutdown_tx,
|
||||
session_lookup: router.session_lookup(),
|
||||
},
|
||||
CdpFrontendReceivers {
|
||||
control_rx,
|
||||
command_rx,
|
||||
},
|
||||
router,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -133,10 +127,6 @@ impl CdpFrontendEndpoint {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn owns_session(&self, frontend_id: u64, session_id: Option<&str>) -> bool {
|
||||
self.session_lookup.owns_session(frontend_id, session_id)
|
||||
}
|
||||
|
||||
pub(crate) async fn command(
|
||||
&self,
|
||||
frontend_id: u64,
|
||||
|
||||
@@ -1,7 +1,4 @@
|
||||
use std::{
|
||||
sync::{Arc, Weak},
|
||||
time::Instant,
|
||||
};
|
||||
use std::time::Instant;
|
||||
|
||||
use anyhow::Result;
|
||||
use moli_protocol::ParsedCdpCommand;
|
||||
@@ -21,32 +18,13 @@ pub(crate) enum CdpPreparedFrontendCommand {
|
||||
}
|
||||
|
||||
pub(crate) struct CdpFrontendRouter {
|
||||
routing: Arc<Mutex<CdpFrontendRoutingState>>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct CdpFrontendSessionLookup {
|
||||
routing: Weak<Mutex<CdpFrontendRoutingState>>,
|
||||
}
|
||||
|
||||
impl CdpFrontendSessionLookup {
|
||||
pub(crate) fn owns_session(&self, frontend_id: u64, session_id: Option<&str>) -> bool {
|
||||
self.routing
|
||||
.upgrade()
|
||||
.is_some_and(|routing| routing.lock().owns_session(frontend_id, session_id))
|
||||
}
|
||||
routing: Mutex<CdpFrontendRoutingState>,
|
||||
}
|
||||
|
||||
impl CdpFrontendRouter {
|
||||
pub(crate) fn new() -> Self {
|
||||
Self {
|
||||
routing: Arc::new(Mutex::new(CdpFrontendRoutingState::default())),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn session_lookup(&self) -> CdpFrontendSessionLookup {
|
||||
CdpFrontendSessionLookup {
|
||||
routing: Arc::downgrade(&self.routing),
|
||||
routing: Mutex::new(CdpFrontendRoutingState::default()),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -67,6 +45,14 @@ impl CdpFrontendRouter {
|
||||
}
|
||||
}
|
||||
|
||||
/// Route a response through the ordinary command table, returning its
|
||||
/// writer when the caller needs a flush confirmation before a side effect.
|
||||
pub(crate) fn enqueue_response(&self, message: Value) -> Option<CdpSocketSink> {
|
||||
let (frontend, message) = self.routing.lock().route_message(message, None)?;
|
||||
let sink = frontend.into_sink();
|
||||
sink.enqueue_owned_message(message).then_some(sink)
|
||||
}
|
||||
|
||||
pub(crate) fn register_browser_frontend(
|
||||
&self,
|
||||
frontend_id: u64,
|
||||
|
||||
@@ -23,6 +23,10 @@ impl CdpRoutedFrontend {
|
||||
pub(super) fn enqueue_message(self, message: Value) -> bool {
|
||||
self.sink.enqueue_owned_message(message)
|
||||
}
|
||||
|
||||
pub(super) fn into_sink(self) -> CdpSocketSink {
|
||||
self.sink
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
@@ -34,13 +38,6 @@ pub(super) struct CdpFrontendRoutingState {
|
||||
}
|
||||
|
||||
impl CdpFrontendRoutingState {
|
||||
pub(super) fn owns_session(&self, frontend_id: u64, session_id: Option<&str>) -> bool {
|
||||
match session_id {
|
||||
Some(session_id) => self.frontends.owns_client_session(frontend_id, session_id),
|
||||
None => self.frontends.base_session_id(frontend_id).is_some(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn register_browser_frontend(
|
||||
&mut self,
|
||||
frontend_id: u64,
|
||||
|
||||
@@ -26,6 +26,51 @@ fn expect_prepared_command(
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn instance_command_responses_use_normal_command_and_session_routes_once() {
|
||||
let router = CdpFrontendRouter::new();
|
||||
let (sink, mut writer) = CdpSocketSink::with_stalled_writer_for_test(4);
|
||||
router
|
||||
.register_browser_frontend(5, "SID-browser".into(), sink)
|
||||
.unwrap();
|
||||
assert!(
|
||||
router.enqueue_protocol_output_sequence(ProtocolOutputSequence::from_messages(vec![
|
||||
json!({
|
||||
"method": "Target.attachedToTarget",
|
||||
"sessionId": "SID-browser",
|
||||
"params": {
|
||||
"sessionId": "SID-child",
|
||||
"targetInfo": {"targetId": "TID-child"},
|
||||
},
|
||||
}),
|
||||
]))
|
||||
);
|
||||
assert_eq!(writer.take_message()["method"], "Target.attachedToTarget");
|
||||
|
||||
for session_id in [None, Some("SID-child")] {
|
||||
let mut request = json!({"id": 42, "method": "Browser.close"});
|
||||
if let Some(session_id) = session_id {
|
||||
request["sessionId"] = json!(session_id);
|
||||
}
|
||||
let command = expect_prepared_command(
|
||||
router.prepare_command(5, ParsedCdpCommand::parse_value(request)),
|
||||
"instance close",
|
||||
);
|
||||
assert_ne!(command.request().id(), 42);
|
||||
let response = json!({"id": command.request().id(), "result": {}});
|
||||
assert!(router.enqueue_response(response.clone()).is_some());
|
||||
let mut expected = json!({"id": 42, "result": {}});
|
||||
if let Some(session_id) = session_id {
|
||||
expected["sessionId"] = json!(session_id);
|
||||
}
|
||||
assert_eq!(writer.take_message(), expected);
|
||||
assert!(
|
||||
router.enqueue_response(response).is_none(),
|
||||
"responding consumes the ordinary pending-command route"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn browser_and_page_client_command_ids_are_isolated_and_restored() {
|
||||
let mut routing = CdpFrontendRoutingState::default();
|
||||
|
||||
@@ -36,6 +36,7 @@ mod actor;
|
||||
mod adapter_scheduler;
|
||||
mod command_dispatch;
|
||||
mod frontend_control;
|
||||
mod instance_command;
|
||||
mod protocol_residence;
|
||||
mod runtime_command_barrier;
|
||||
mod runtime_dispatch;
|
||||
|
||||
@@ -377,6 +377,7 @@ async fn run_cdp_scheduler_actor(
|
||||
&mut in_flight_commands,
|
||||
&mut blocked_commands,
|
||||
&mut next_in_flight_command_token,
|
||||
owner_lifecycle.as_ref(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
@@ -1408,6 +1409,7 @@ async fn handle_frontend_command(
|
||||
in_flight_commands: &mut InFlightCommands,
|
||||
blocked_commands: &mut VecDeque<BlockedCommandDispatch>,
|
||||
next_in_flight_command_token: &mut u64,
|
||||
owner_lifecycle: Option<&CdpOwnerActorLifecycle>,
|
||||
) -> bool {
|
||||
let CdpFrontendCommand {
|
||||
frontend_id,
|
||||
@@ -1427,6 +1429,16 @@ async fn handle_frontend_command(
|
||||
}
|
||||
};
|
||||
trace_command(&command, "frontend_command_received", None, None, None);
|
||||
if let Some(owner_lifecycle) = owner_lifecycle
|
||||
&& super::instance_command::handle_instance_command(
|
||||
&command,
|
||||
frontend_router,
|
||||
&owner_lifecycle.instance_control,
|
||||
)
|
||||
.await
|
||||
{
|
||||
return true;
|
||||
}
|
||||
handle_client_command_with_interleaved_output(
|
||||
frontend_router,
|
||||
scheduler,
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
use moli_cookie_jar::StoredCookie;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
use crate::{cdp_frontend::CdpFrontendControlRequest, cdp_frontend_router::CdpFrontendRouter};
|
||||
use crate::{
|
||||
BrowserInstanceControl, cdp_frontend::CdpFrontendControlRequest,
|
||||
cdp_frontend_router::CdpFrontendRouter,
|
||||
};
|
||||
|
||||
use self::target_control::CdpFrontendTargetControl;
|
||||
use super::CdpScheduler;
|
||||
@@ -27,6 +30,7 @@ impl CdpCookieSnapshot {
|
||||
|
||||
pub(crate) struct CdpOwnerActorLifecycle {
|
||||
pub(crate) checkpoint_tx: mpsc::UnboundedSender<CdpCookieSnapshot>,
|
||||
pub(crate) instance_control: BrowserInstanceControl,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
//! CDP instance commands run after normal frontend routing and before page work.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use moli_protocol::ParsedCdpCommand;
|
||||
use serde_json::json;
|
||||
|
||||
use crate::{BrowserInstanceControl, InstanceCommand, cdp_frontend_router::CdpFrontendRouter};
|
||||
|
||||
const BROWSER_CLOSE_FLUSH_BUDGET: Duration = Duration::from_secs(1);
|
||||
|
||||
/// The actor has already validated the request's session through the Router.
|
||||
/// Its internal command ID routes the response back to the original frontend,
|
||||
/// client command ID, and public session using the ordinary response path.
|
||||
pub(super) async fn handle_instance_command(
|
||||
command: &ParsedCdpCommand,
|
||||
router: &CdpFrontendRouter,
|
||||
control: &BrowserInstanceControl,
|
||||
) -> bool {
|
||||
let instance_command = match command.method() {
|
||||
"Browser.close" => InstanceCommand::Shutdown,
|
||||
_ => return false,
|
||||
};
|
||||
let Some(sink) = router.enqueue_response(json!({"id": command.request().id(), "result": {}}))
|
||||
else {
|
||||
return true;
|
||||
};
|
||||
if matches!(
|
||||
tokio::time::timeout(BROWSER_CLOSE_FLUSH_BUDGET, sink.flush()).await,
|
||||
Ok(true)
|
||||
) {
|
||||
control.request(instance_command);
|
||||
} else {
|
||||
// A failed acknowledgment retires this frontend, not the instance or
|
||||
// the actor serving other connections.
|
||||
sink.close();
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::{cdp_frontend_router::CdpPreparedFrontendCommand, cdp_writer::CdpSocketSink};
|
||||
|
||||
#[tokio::test]
|
||||
async fn failed_or_unconfirmed_flush_does_not_stop_the_instance() {
|
||||
let control = BrowserInstanceControl::new();
|
||||
let router = CdpFrontendRouter::new();
|
||||
let (stalled_sink, stalled_writer) = CdpSocketSink::with_stalled_writer_for_test(2);
|
||||
router
|
||||
.register_browser_frontend(1, "hidden-dead".into(), CdpSocketSink::for_test())
|
||||
.unwrap();
|
||||
router
|
||||
.register_browser_frontend(2, "hidden-stalled".into(), stalled_sink)
|
||||
.unwrap();
|
||||
for frontend_id in [1, 2] {
|
||||
let parsed = ParsedCdpCommand::parse_value(json!({
|
||||
"id": 42, "method": "Browser.close",
|
||||
}));
|
||||
let Some(CdpPreparedFrontendCommand::Command(command)) =
|
||||
router.prepare_command(frontend_id, parsed)
|
||||
else {
|
||||
panic!("registered frontend must route normally");
|
||||
};
|
||||
assert!(handle_instance_command(&command, &router, &control).await);
|
||||
}
|
||||
assert!(
|
||||
!stalled_writer.is_open(),
|
||||
"failed flush closes its frontend"
|
||||
);
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(10), control.shutdown_requested())
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -1,156 +0,0 @@
|
||||
//! Translate CDP instance commands without entering a page's work queue.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use moli_protocol::ParsedCdpCommand;
|
||||
use serde_json::{Value, json};
|
||||
|
||||
use super::{BrowserInstanceControl, InstanceCommand};
|
||||
use crate::{cdp_frontend::CdpFrontendEndpoint, cdp_writer::CdpSocketSink};
|
||||
|
||||
const BROWSER_CLOSE_FLUSH_BUDGET: Duration = Duration::from_secs(1);
|
||||
|
||||
/// Return true when this adapter consumed a command and the frontend should
|
||||
/// close, including when its acknowledgment could not be flushed.
|
||||
pub(super) async fn handle_instance_command(
|
||||
command: &ParsedCdpCommand,
|
||||
endpoint: &CdpFrontendEndpoint,
|
||||
frontend_id: u64,
|
||||
sink: &CdpSocketSink,
|
||||
control: &BrowserInstanceControl,
|
||||
) -> bool {
|
||||
let Some((command, response)) = instance_command(command, endpoint, frontend_id) else {
|
||||
return false;
|
||||
};
|
||||
acknowledge_instance_command(sink, control, command, response).await;
|
||||
true
|
||||
}
|
||||
|
||||
// Instance control bypasses page work, using the normal wire parser and the
|
||||
// owner's session registry. Chromium permits it on page and attached routes too.
|
||||
fn instance_command(
|
||||
command: &ParsedCdpCommand,
|
||||
endpoint: &CdpFrontendEndpoint,
|
||||
frontend_id: u64,
|
||||
) -> Option<(InstanceCommand, Value)> {
|
||||
let request = command.request();
|
||||
if request.method() != "Browser.close"
|
||||
|| !endpoint.owns_session(frontend_id, request.session_id())
|
||||
{
|
||||
return None;
|
||||
}
|
||||
let mut response = json!({"id": request.id(), "result": {}});
|
||||
if let Some(session_id) = request.session_id() {
|
||||
response["sessionId"] = json!(session_id);
|
||||
}
|
||||
Some((InstanceCommand::Shutdown, response))
|
||||
}
|
||||
|
||||
async fn acknowledge_instance_command(
|
||||
sink: &CdpSocketSink,
|
||||
control: &BrowserInstanceControl,
|
||||
command: InstanceCommand,
|
||||
response: Value,
|
||||
) -> bool {
|
||||
if !sink.enqueue_owned_message(response) {
|
||||
return false;
|
||||
}
|
||||
if !matches!(
|
||||
tokio::time::timeout(BROWSER_CLOSE_FLUSH_BUDGET, sink.flush()).await,
|
||||
Ok(true)
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
control.request(command);
|
||||
true
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn only_a_valid_owned_frontend_close_can_request_instance_shutdown() {
|
||||
let (endpoint, _receivers, router) = crate::cdp_frontend::cdp_frontend_channel();
|
||||
router
|
||||
.register_browser_frontend(1, "hidden-browser".into(), CdpSocketSink::for_test())
|
||||
.unwrap();
|
||||
router
|
||||
.register_target_frontend(
|
||||
2,
|
||||
"page".into(),
|
||||
"hidden-page".into(),
|
||||
CdpSocketSink::for_test(),
|
||||
)
|
||||
.unwrap();
|
||||
for value in [
|
||||
json!({"id": 7, "method": "Browser.close"}),
|
||||
json!({"id": 7, "method": "Browser.close", "params": {}}),
|
||||
json!({"id": 7, "method": "Browser.close", "params": {"ignored": true}}),
|
||||
json!({"id": 7, "method": "Browser.close", "params": []}),
|
||||
json!({"id": 7, "method": "Browser.close", "params": null}),
|
||||
] {
|
||||
let command = ParsedCdpCommand::parse_value(value).unwrap();
|
||||
for frontend_id in [1, 2] {
|
||||
assert_eq!(
|
||||
instance_command(&command, &endpoint, frontend_id),
|
||||
Some((InstanceCommand::Shutdown, json!({"id": 7, "result": {}})))
|
||||
);
|
||||
}
|
||||
}
|
||||
for value in [
|
||||
json!({"id": 7, "method": "Browser.close", "sessionId": "unknown"}),
|
||||
json!({"id": 7, "method": "Browser.close", "sessionId": "hidden-browser"}),
|
||||
json!({"id": 7, "method": "browser.close"}),
|
||||
] {
|
||||
let command = ParsedCdpCommand::parse_value(value).unwrap();
|
||||
assert!(instance_command(&command, &endpoint, 1).is_none());
|
||||
}
|
||||
for value in [
|
||||
json!({"id": 7, "method": "Browser.close", "sessionId": null}),
|
||||
json!({"id": -1, "method": "Browser.close"}),
|
||||
json!({"id": "7", "method": "Browser.close"}),
|
||||
json!({"method": "Browser.close"}),
|
||||
] {
|
||||
assert!(ParsedCdpCommand::parse_value(value).is_err());
|
||||
}
|
||||
drop(router);
|
||||
let command =
|
||||
ParsedCdpCommand::parse_value(json!({"id": 7, "method": "Browser.close"})).unwrap();
|
||||
assert!(
|
||||
instance_command(&command, &endpoint, 1).is_none(),
|
||||
"a transport must not keep its departed owner alive"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn failed_or_unconfirmed_flush_does_not_stop_the_instance() {
|
||||
let control = BrowserInstanceControl::new();
|
||||
let sink = CdpSocketSink::for_test();
|
||||
assert!(
|
||||
!acknowledge_instance_command(
|
||||
&sink,
|
||||
&control,
|
||||
InstanceCommand::Shutdown,
|
||||
json!({"id": 1, "result": {}})
|
||||
)
|
||||
.await
|
||||
);
|
||||
let (sink, _stalled_writer) = CdpSocketSink::with_stalled_writer_for_test(2);
|
||||
assert!(
|
||||
!acknowledge_instance_command(
|
||||
&sink,
|
||||
&control,
|
||||
InstanceCommand::Shutdown,
|
||||
json!({"id": 2, "result": {}})
|
||||
)
|
||||
.await
|
||||
);
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(10), control.shutdown_requested())
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
assert!(control.connection().is_some());
|
||||
}
|
||||
}
|
||||
@@ -19,7 +19,8 @@ use crate::{
|
||||
};
|
||||
|
||||
use super::{
|
||||
CookieProfileCommit, SharedCookieProfile, cdp_agent_host::SharedCdpAgentHostDirectory,
|
||||
BrowserInstanceControl, CookieProfileCommit, SharedCookieProfile,
|
||||
cdp_agent_host::SharedCdpAgentHostDirectory,
|
||||
protocol_local_executor::spawn_protocol_local_task,
|
||||
};
|
||||
|
||||
@@ -58,9 +59,11 @@ struct CdpOwnerRuntimeConfig {
|
||||
storage_partition: Arc<StoragePartitionState>,
|
||||
navigation_runtime_config: NavigationRuntimeConfig,
|
||||
screencast_interval_ms: u32,
|
||||
instance_control: BrowserInstanceControl,
|
||||
}
|
||||
|
||||
impl SharedCdpOwnerRegistry {
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub(super) fn new(
|
||||
directory: SharedCdpAgentHostDirectory,
|
||||
target_id_allocator: Arc<AtomicU64>,
|
||||
@@ -69,6 +72,7 @@ impl SharedCdpOwnerRegistry {
|
||||
storage_partition: Arc<StoragePartitionState>,
|
||||
navigation_runtime_config: NavigationRuntimeConfig,
|
||||
screencast_interval_ms: u32,
|
||||
instance_control: BrowserInstanceControl,
|
||||
) -> Self {
|
||||
Self {
|
||||
inner: Arc::new(CdpOwnerRegistryInner {
|
||||
@@ -80,6 +84,7 @@ impl SharedCdpOwnerRegistry {
|
||||
storage_partition,
|
||||
navigation_runtime_config,
|
||||
screencast_interval_ms,
|
||||
instance_control,
|
||||
},
|
||||
state: Mutex::new(CdpOwnerRegistryState::default()),
|
||||
shared_owner: Mutex::new(None),
|
||||
@@ -102,7 +107,7 @@ impl SharedCdpOwnerRegistry {
|
||||
|
||||
fn spawn_owner(&self) -> Result<(u64, CdpFrontendEndpoint)> {
|
||||
let owner_id = self.inner.config.directory.allocate_owner_id();
|
||||
let (endpoint, receivers, frontend_router) = cdp_frontend_channel();
|
||||
let (endpoint, receivers) = cdp_frontend_channel();
|
||||
let lifecycle_observer = self
|
||||
.inner
|
||||
.config
|
||||
@@ -134,7 +139,6 @@ impl SharedCdpOwnerRegistry {
|
||||
);
|
||||
let owner_finished_rx = spawn_owner_task(
|
||||
receivers,
|
||||
frontend_router,
|
||||
initial_storage_partition,
|
||||
self.inner.config.navigation_runtime_config.clone(),
|
||||
self.inner.config.screencast_interval_ms,
|
||||
@@ -145,6 +149,7 @@ impl SharedCdpOwnerRegistry {
|
||||
)),
|
||||
Some(CdpOwnerActorLifecycle {
|
||||
checkpoint_tx: checkpoint_tx.clone(),
|
||||
instance_control: self.inner.config.instance_control.clone(),
|
||||
}),
|
||||
checkpoint_tx,
|
||||
);
|
||||
@@ -203,7 +208,6 @@ impl Drop for CdpOwnerRegistryInner {
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn spawn_owner_task(
|
||||
receivers: CdpFrontendReceivers,
|
||||
frontend_router: CdpFrontendRouter,
|
||||
initial_storage_partition: CdpInitialStoragePartition,
|
||||
navigation_runtime_config: NavigationRuntimeConfig,
|
||||
screencast_interval_ms: u32,
|
||||
@@ -222,7 +226,7 @@ fn spawn_owner_task(
|
||||
let actor = spawn_cdp_scheduler_actor(
|
||||
scheduler,
|
||||
scheduler_receivers,
|
||||
frontend_router,
|
||||
CdpFrontendRouter::new(),
|
||||
receivers,
|
||||
owner_lifecycle,
|
||||
);
|
||||
|
||||
@@ -3,7 +3,7 @@ use futures_util::StreamExt;
|
||||
use moli_protocol::ParsedCdpCommand;
|
||||
use std::time::Duration;
|
||||
|
||||
use super::{BrowserInstanceControl, cdp_instance_control::handle_instance_command};
|
||||
use super::BrowserInstanceControl;
|
||||
use crate::{cdp_frontend::CdpFrontendEndpoint, cdp_writer::spawn_socket_sink};
|
||||
|
||||
pub(super) enum CdpFrontendSocketKind {
|
||||
@@ -57,13 +57,6 @@ pub(super) async fn run_cdp_frontend_socket(
|
||||
match message {
|
||||
Message::Text(text) => {
|
||||
let command = ParsedCdpCommand::parse_str(text.to_string());
|
||||
if let Ok(command) = &command
|
||||
&& handle_instance_command(
|
||||
command, &endpoint, frontend_id, &socket_sink, &control,
|
||||
).await
|
||||
{
|
||||
break CdpSocketExit::TransportClose;
|
||||
}
|
||||
if !endpoint.command(frontend_id, command).await {
|
||||
break CdpSocketExit::TransportClose;
|
||||
}
|
||||
|
||||
@@ -41,7 +41,6 @@ pub use instance_control::{BrowserInstanceControl, InstanceCommand};
|
||||
|
||||
mod cdp;
|
||||
mod cdp_agent_host;
|
||||
mod cdp_instance_control;
|
||||
mod cdp_owner;
|
||||
mod cdp_socket;
|
||||
mod instance_control;
|
||||
@@ -164,13 +163,13 @@ impl ProtocolServer {
|
||||
|
||||
info!(%addr, "protocol server listening");
|
||||
|
||||
let mut app_state = AppState::new_with_storage_partition_and_runtime_config(
|
||||
let app_state = AppState::new_with_storage_partition_and_runtime_config(
|
||||
addr,
|
||||
self.storage_partition.clone(),
|
||||
self.navigation_runtime_config.clone(),
|
||||
self.config.screencast_interval_ms,
|
||||
control,
|
||||
)?;
|
||||
app_state.control = control;
|
||||
|
||||
serve_listener(listener, app_state).await
|
||||
}
|
||||
@@ -830,6 +829,7 @@ impl AppState {
|
||||
LayoutPolicy::default(),
|
||||
),
|
||||
crate::config::DEFAULT_SCREENCAST_INTERVAL_MS,
|
||||
BrowserInstanceControl::new(),
|
||||
)
|
||||
}
|
||||
|
||||
@@ -838,6 +838,7 @@ impl AppState {
|
||||
storage_partition: Arc<StoragePartitionState>,
|
||||
navigation_runtime_config: NavigationRuntimeConfig,
|
||||
screencast_interval_ms: u32,
|
||||
control: BrowserInstanceControl,
|
||||
) -> anyhow::Result<Self> {
|
||||
Ok(Self::from_parts(
|
||||
addr,
|
||||
@@ -845,6 +846,7 @@ impl AppState {
|
||||
storage_partition,
|
||||
navigation_runtime_config,
|
||||
screencast_interval_ms,
|
||||
control,
|
||||
))
|
||||
}
|
||||
|
||||
@@ -854,6 +856,7 @@ impl AppState {
|
||||
storage_partition: Arc<StoragePartitionState>,
|
||||
navigation_runtime_config: NavigationRuntimeConfig,
|
||||
screencast_interval_ms: u32,
|
||||
control: BrowserInstanceControl,
|
||||
) -> Self {
|
||||
let cdp_agent_host_directory = SharedCdpAgentHostDirectory::default();
|
||||
let cdp_target_id_allocator = Arc::new(AtomicU64::new(0));
|
||||
@@ -866,9 +869,10 @@ impl AppState {
|
||||
storage_partition.clone(),
|
||||
navigation_runtime_config.clone(),
|
||||
screencast_interval_ms,
|
||||
control.clone(),
|
||||
);
|
||||
Self {
|
||||
control: BrowserInstanceControl::new(),
|
||||
control,
|
||||
browser_ws_url: format!("ws://{addr}/devtools/browser/{DEFAULT_BROWSER_ID}"),
|
||||
page_ws_url: format!("ws://{addr}/devtools/page/{DEFAULT_TARGET_ID}"),
|
||||
bidi_ws_url: format!("ws://{addr}/session"),
|
||||
|
||||
@@ -20,6 +20,7 @@ async fn instance(
|
||||
OptionalResourceFetchMask::NONE,
|
||||
),
|
||||
crate::config::DEFAULT_SCREENCAST_INTERVAL_MS,
|
||||
BrowserInstanceControl::new(),
|
||||
);
|
||||
let serving = state.clone();
|
||||
let task = tokio::spawn(async move { serve_listener(listener, serving).await });
|
||||
|
||||
@@ -11,6 +11,7 @@ async fn webdriver_classic_hidden_scrollbars_preserve_server_configuration() {
|
||||
)
|
||||
.with_scrollbars_hidden(true),
|
||||
crate::config::DEFAULT_SCREENCAST_INTERVAL_MS,
|
||||
BrowserInstanceControl::new(),
|
||||
)
|
||||
.unwrap();
|
||||
let app = build_router(state);
|
||||
@@ -2831,6 +2832,7 @@ async fn webdriver_classic_click_respects_dom_first_and_real_layout_policies() {
|
||||
policy,
|
||||
),
|
||||
crate::config::DEFAULT_SCREENCAST_INTERVAL_MS,
|
||||
BrowserInstanceControl::new(),
|
||||
)
|
||||
.unwrap();
|
||||
let app = build_router(state);
|
||||
|
||||
@@ -276,6 +276,7 @@ fn protocol_server_test_state(
|
||||
optional_resource_fetch_mask,
|
||||
),
|
||||
crate::config::DEFAULT_SCREENCAST_INTERVAL_MS,
|
||||
BrowserInstanceControl::new(),
|
||||
)
|
||||
.expect("test app state should initialize")
|
||||
}
|
||||
@@ -437,6 +438,7 @@ async fn spawn_test_protocol_server_with_runtime_config(
|
||||
storage_partition,
|
||||
navigation_runtime_config,
|
||||
crate::config::DEFAULT_SCREENCAST_INTERVAL_MS,
|
||||
BrowserInstanceControl::new(),
|
||||
)
|
||||
.expect("test app state should initialize");
|
||||
let server = tokio::spawn(async move {
|
||||
|
||||
Reference in New Issue
Block a user