use core::convert::Infallible as Never;
use std::cell::RefCell;
use std::collections::HashMap;
use std::fmt::Debug;
use std::net::SocketAddr;
use std::pin::pin;
use std::process;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::OnceLock;
use std::task::Poll;
use std::thread;
use deno_core::InspectorMsg;
use deno_core::InspectorSessionChannels;
use deno_core::InspectorSessionKind;
use deno_core::InspectorSessionProxy;
use deno_core::JsRuntimeInspector;
use deno_core::futures::channel::mpsc;
use deno_core::futures::channel::mpsc::UnboundedReceiver;
use deno_core::futures::channel::mpsc::UnboundedSender;
use deno_core::futures::channel::oneshot;
use deno_core::futures::future;
use deno_core::futures::prelude::*;
use deno_core::futures::stream::StreamExt;
use deno_core::parking_lot::Mutex;
use deno_core::serde_json;
use deno_core::serde_json::Value;
use deno_core::serde_json::json;
use deno_core::unsync::spawn;
use deno_core::url::Url;
use fastwebsockets::Frame;
use fastwebsockets::OpCode;
use fastwebsockets::WebSocket;
use hyper::body::Bytes;
use hyper_util::rt::TokioIo;
use tokio::net::TcpListener;
use tokio::sync::broadcast;
use uuid::Uuid;
pub struct InspectorServerUrl(pub String);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct InspectPublishUid {
pub console: bool,
pub http: bool,
}
impl Default for InspectPublishUid {
fn default() -> Self {
Self {
console: true,
http: true,
}
}
}
pub struct InspectorServer {
pub host: SocketAddr,
register_inspector_tx: UnboundedSender<InspectorInfo>,
shutdown_server_tx: Option<broadcast::Sender<()>>,
reset_tx: broadcast::Sender<()>,
thread_handle: Mutex<Option<thread::JoinHandle<()>>>,
}
static GLOBAL_INSPECTOR_SERVER: OnceLock<Mutex<Option<Arc<InspectorServer>>>> =
OnceLock::new();
fn global_server() -> &'static Mutex<Option<Arc<InspectorServer>>> {
GLOBAL_INSPECTOR_SERVER.get_or_init(|| Mutex::new(None))
}
pub fn get_inspector_server() -> Option<Arc<InspectorServer>> {
global_server().lock().clone()
}
pub fn stop_inspector_server() {
if let Some(server) = global_server().lock().take() {
server.stop();
}
}
pub fn create_inspector_server(
host: SocketAddr,
name: &'static str,
publish_uid: InspectPublishUid,
) -> Result<Arc<InspectorServer>, InspectorServerError> {
let mut guard = global_server().lock();
if let Some(server) = guard.as_ref() {
return Ok(server.clone());
}
let server = Arc::new(InspectorServer::new(host, name, publish_uid)?);
*guard = Some(server.clone());
Ok(server)
}
static RESTART_NOTIFIER: OnceLock<broadcast::Sender<()>> = OnceLock::new();
fn get_restart_notifier() -> broadcast::Sender<()> {
RESTART_NOTIFIER
.get_or_init(|| broadcast::channel(16).0)
.clone()
}
pub fn notify_restart() {
let sender = get_restart_notifier();
let _ = sender.send(()); }
#[derive(Debug, thiserror::Error, deno_error::JsError)]
pub enum InspectorServerError {
#[class(inherit)]
#[error(transparent)]
Io(#[from] std::io::Error),
#[class(inherit)]
#[error("Failed to start inspector server at \"{host}\"")]
Connect {
host: SocketAddr,
#[source]
#[inherit]
source: std::io::Error,
},
#[class(inherit)]
#[error("Failed to get inspector server's assigned address")]
LocalAddr {
host: SocketAddr,
#[source]
#[inherit]
source: std::io::Error,
},
}
fn create_basic_runtime() -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_current_thread()
.enable_io()
.enable_time()
.max_blocking_threads(4)
.build()
.unwrap()
}
impl InspectorServer {
pub fn new(
host: SocketAddr,
name: &'static str,
publish_uid: InspectPublishUid,
) -> Result<Self, InspectorServerError> {
let (register_inspector_tx, register_inspector_rx) =
mpsc::unbounded::<InspectorInfo>();
let (shutdown_server_tx, shutdown_server_rx) = broadcast::channel(1);
let (reset_tx, reset_rx) = broadcast::channel(1);
let tcp_listener = std::net::TcpListener::bind(host)
.map_err(|source| InspectorServerError::Connect { host, source })?;
tcp_listener.set_nonblocking(true)?;
let host = tcp_listener
.local_addr()
.map_err(|source| InspectorServerError::LocalAddr { host, source })?;
let thread_handle = thread::spawn(move || {
let rt = create_basic_runtime();
let local = tokio::task::LocalSet::new();
local.block_on(
&rt,
server(
tcp_listener,
register_inspector_rx,
shutdown_server_rx,
reset_rx,
name,
publish_uid,
),
)
});
Ok(Self {
host,
register_inspector_tx,
shutdown_server_tx: Some(shutdown_server_tx),
reset_tx,
thread_handle: Mutex::new(Some(thread_handle)),
})
}
pub fn register_inspector(
&self,
module_url: String,
inspector: Rc<JsRuntimeInspector>,
wait_for_session: bool,
) -> InspectorServerUrl {
let session_sender = inspector.get_session_sender();
let deregister_rx = inspector.add_deregister_handler();
let info = InspectorInfo::new(
self.host,
session_sender,
deregister_rx,
module_url,
wait_for_session,
);
let url = InspectorServerUrl(
info.get_websocket_debugger_url(&self.host.to_string()),
);
self.register_inspector_tx.unbounded_send(info).unwrap();
url
}
pub fn stop(&self) {
let _ = self.reset_tx.send(());
}
}
impl Drop for InspectorServer {
fn drop(&mut self) {
if let Some(shutdown_server_tx) = self.shutdown_server_tx.take() {
shutdown_server_tx
.send(())
.expect("unable to send shutdown signal");
}
if let Some(thread_handle) = self.thread_handle.lock().take() {
thread_handle.join().expect("unable to join thread");
}
}
}
fn handle_ws_request(
req: http::Request<hyper::body::Incoming>,
inspector_map_rc: Rc<RefCell<HashMap<Uuid, InspectorInfo>>>,
) -> http::Result<http::Response<Box<http_body_util::Full<Bytes>>>> {
let (parts, body) = req.into_parts();
let req = http::Request::from_parts(parts, ());
let maybe_uuid = req
.uri()
.path()
.strip_prefix("/ws/")
.and_then(|s| Uuid::parse_str(s).ok());
let Some(uuid) = maybe_uuid else {
return http::Response::builder()
.status(http::StatusCode::BAD_REQUEST)
.body(Box::new(Bytes::from("Malformed inspector UUID").into()));
};
let new_session_tx = {
let inspector_map = inspector_map_rc.borrow();
let maybe_inspector_info = inspector_map.get(&uuid);
if maybe_inspector_info.is_none() {
return http::Response::builder()
.status(http::StatusCode::NOT_FOUND)
.body(Box::new(Bytes::from("Invalid inspector UUID").into()));
}
let info = maybe_inspector_info.unwrap();
info.new_session_tx.clone()
};
let (parts, _) = req.into_parts();
let mut req = http::Request::from_parts(parts, body);
let Ok((resp, upgrade_fut)) = fastwebsockets::upgrade::upgrade(&mut req)
else {
return http::Response::builder()
.status(http::StatusCode::BAD_REQUEST)
.body(Box::new(
Bytes::from("Not a valid Websocket Request").into(),
));
};
spawn(async move {
let websocket = match upgrade_fut.await {
Ok(w) => w,
Err(err) => {
log::error!(
"Inspector server failed to upgrade to WS connection: {:?}",
err
);
return;
}
};
let (outbound_tx, outbound_rx) = mpsc::unbounded();
let (inbound_tx, inbound_rx) = mpsc::unbounded();
let inspector_session_proxy = InspectorSessionProxy {
channels: InspectorSessionChannels::Regular {
tx: outbound_tx,
rx: inbound_rx,
},
kind: InspectorSessionKind::NonBlocking {
wait_for_disconnect: true,
},
};
log::info!("Debugger session started.");
let _ = new_session_tx.unbounded_send(inspector_session_proxy);
pump_websocket_messages(websocket, inbound_tx, outbound_rx).await;
});
let (parts, _body) = resp.into_parts();
let resp = http::Response::from_parts(
parts,
Box::new(http_body_util::Full::new(Bytes::new())),
);
Ok(resp)
}
fn handle_json_request(
inspector_map: Rc<RefCell<HashMap<Uuid, InspectorInfo>>>,
host: Option<String>,
) -> http::Result<http::Response<Box<http_body_util::Full<Bytes>>>> {
let data = inspector_map
.borrow()
.values()
.map(move |info| info.get_json_metadata(&host))
.collect::<Vec<_>>();
let body: http_body_util::Full<Bytes> =
Bytes::from(serde_json::to_string(&data).unwrap()).into();
http::Response::builder()
.status(http::StatusCode::OK)
.header(http::header::CONTENT_TYPE, "application/json")
.body(Box::new(body))
}
fn handle_json_version_request(
version_response: Value,
) -> http::Result<http::Response<Box<http_body_util::Full<Bytes>>>> {
let body = Box::new(http_body_util::Full::from(
serde_json::to_string(&version_response).unwrap(),
));
http::Response::builder()
.status(http::StatusCode::OK)
.header(http::header::CONTENT_TYPE, "application/json")
.body(body)
}
fn handle_ws_events_request(
req: http::Request<hyper::body::Incoming>,
) -> http::Result<http::Response<Box<http_body_util::Full<Bytes>>>> {
#[allow(clippy::disallowed_methods, reason = "TODO: pass a sys in here")]
if std::env::var("UNSTABLE_INSPECTOR_WS_EVENTS").is_err() {
return http::Response::builder()
.status(http::StatusCode::NOT_FOUND)
.body(Box::new(http_body_util::Full::new(Bytes::from(
"Not Found",
))));
}
let (parts, body) = req.into_parts();
let req = http::Request::from_parts(parts, ());
let (parts, _) = req.into_parts();
let mut req = http::Request::from_parts(parts, body);
let Ok((resp, upgrade_fut)) = fastwebsockets::upgrade::upgrade(&mut req)
else {
return http::Response::builder()
.status(http::StatusCode::BAD_REQUEST)
.body(Box::new(http_body_util::Full::new(Bytes::from(
"Not a valid Websocket Request",
))));
};
let restart_rx = get_restart_notifier().subscribe();
spawn(async move {
let websocket = match upgrade_fut.await {
Ok(w) => w,
Err(err) => {
log::error!(
"Inspector server failed to upgrade to WS connection for /ws/events: {:?}",
err
);
return;
}
};
log::debug!("Deno event session started.");
pump_event_notifications(websocket, restart_rx).await;
});
let (parts, _body) = resp.into_parts();
let resp = http::Response::from_parts(
parts,
Box::new(http_body_util::Full::new(Bytes::new())),
);
Ok(resp)
}
async fn pump_event_notifications(
mut websocket: WebSocket<TokioIo<hyper::upgrade::Upgraded>>,
mut restart_rx: broadcast::Receiver<()>,
) {
loop {
tokio::select! {
result = restart_rx.recv() => {
match result {
Ok(()) => {
#[allow(clippy::disallowed_methods, reason = "TODO: pass a sys in here")]
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let msg = json!({
"type": "restart",
"timestamp": timestamp,
});
let frame = Frame::text(msg.to_string().into_bytes().into());
if websocket.write_frame(frame).await.is_err() {
break;
}
}
Err(broadcast::error::RecvError::Lagged(_)) => {
continue;
}
Err(broadcast::error::RecvError::Closed) => {
break;
}
}
}
result = websocket.read_frame() => {
match result {
Ok(frame) => {
if frame.opcode == OpCode::Close {
log::debug!("Deno event session ended");
break;
}
}
Err(_) => {
break;
}
}
}
}
}
}
async fn server(
listener: std::net::TcpListener,
register_inspector_rx: UnboundedReceiver<InspectorInfo>,
shutdown_server_rx: broadcast::Receiver<()>,
reset_rx: broadcast::Receiver<()>,
name: &str,
publish_uid: InspectPublishUid,
) {
let inspector_map_ =
Rc::new(RefCell::new(HashMap::<Uuid, InspectorInfo>::new()));
let inspector_map = Rc::clone(&inspector_map_);
let register_inspector_handler = listen_for_new_inspectors(
register_inspector_rx,
inspector_map.clone(),
publish_uid,
)
.boxed_local();
let inspector_map = Rc::clone(&inspector_map_);
let mut reset_rx_deregister = reset_rx.resubscribe();
let deregister_inspector_handler = future::poll_fn(|cx| {
if let Ok(()) = reset_rx_deregister.try_recv() {
inspector_map.borrow_mut().clear();
}
inspector_map
.borrow_mut()
.retain(|_, info| info.deregister_rx.poll_unpin(cx) == Poll::Pending);
Poll::<Never>::Pending
})
.boxed_local();
let json_version_response = json!({
"Browser": name,
"Protocol-Version": "1.3",
"V8-Version": deno_core::v8::VERSION_STRING,
});
let mut reset_rx_server = reset_rx.resubscribe();
let listener = match TcpListener::from_std(listener) {
Ok(l) => l,
Err(err) => {
log::error!("Cannot start inspector server: {:?}", err);
return;
}
};
let server_handler = async move {
loop {
let mut rx = shutdown_server_rx.resubscribe();
let mut shutdown_rx = pin!(rx.recv());
let mut accept = pin!(listener.accept());
let stream = tokio::select! {
accept_result = &mut accept => {
match accept_result {
Ok((s, _)) => s,
Err(err) => {
log::error!("Failed to accept inspector connection: {:?}", err);
continue;
}
}
},
_ = &mut shutdown_rx => {
break;
}
};
let io = TokioIo::new(stream);
let inspector_map = Rc::clone(&inspector_map_);
let json_version_response = json_version_response.clone();
let mut shutdown_server_rx = shutdown_server_rx.resubscribe();
let mut reset_rx_conn = reset_rx.resubscribe();
let service = hyper::service::service_fn(
move |req: http::Request<hyper::body::Incoming>| {
future::ready({
let host = req
.headers()
.get("host")
.and_then(|host| host.to_str().ok())
.and_then(|host| Url::parse(&format!("http://{host}")).ok())
.and_then(|url| match (url.host(), url.port()) {
(Some(host), Some(port)) => Some(format!("{host}:{port}")),
(Some(host), None) => Some(format!("{host}")),
_ => None,
});
match (req.method(), req.uri().path()) {
(&http::Method::GET, "/ws/events") => {
handle_ws_events_request(req)
}
(&http::Method::GET, path) if path.starts_with("/ws/") => {
handle_ws_request(req, Rc::clone(&inspector_map))
}
(&http::Method::GET, "/json/version") if publish_uid.http => {
handle_json_version_request(json_version_response.clone())
}
(&http::Method::GET, "/json") if publish_uid.http => {
handle_json_request(Rc::clone(&inspector_map), host)
}
(&http::Method::GET, "/json/list") if publish_uid.http => {
handle_json_request(Rc::clone(&inspector_map), host)
}
_ => http::Response::builder()
.status(http::StatusCode::NOT_FOUND)
.body(Box::new(http_body_util::Full::new(Bytes::from(
"Not Found",
)))),
}
})
},
);
deno_core::unsync::spawn(async move {
let server = hyper::server::conn::http1::Builder::new();
let mut conn =
pin!(server.serve_connection(io, service).with_upgrades());
let mut shutdown_rx = pin!(shutdown_server_rx.recv());
let mut reset_rx = pin!(reset_rx_conn.recv());
tokio::select! {
result = conn.as_mut() => {
if let Err(err) = result {
log::error!("Failed to serve connection: {:?}", err);
}
},
_ = &mut shutdown_rx => {
conn.as_mut().graceful_shutdown();
let _ = conn.await;
},
_ = &mut reset_rx => {
}
}
});
}
}
.boxed_local();
tokio::select! {
_ = register_inspector_handler => {},
_ = deregister_inspector_handler => unreachable!(),
_ = server_handler => {},
_ = reset_rx_server.recv() => {},
}
}
async fn listen_for_new_inspectors(
mut register_inspector_rx: UnboundedReceiver<InspectorInfo>,
inspector_map: Rc<RefCell<HashMap<Uuid, InspectorInfo>>>,
publish_uid: InspectPublishUid,
) {
while let Some(info) = register_inspector_rx.next().await {
if publish_uid.console {
log::info!(
"Debugger listening on {}",
info.get_websocket_debugger_url(&info.host.to_string())
);
log::info!("Visit chrome://inspect to connect to the debugger.");
if info.wait_for_session {
log::info!("Deno is waiting for debugger to connect.");
}
}
if inspector_map.borrow_mut().insert(info.uuid, info).is_some() {
panic!("Inspector UUID already in map");
}
}
}
async fn pump_websocket_messages(
mut websocket: WebSocket<TokioIo<hyper::upgrade::Upgraded>>,
inbound_tx: UnboundedSender<String>,
mut outbound_rx: UnboundedReceiver<InspectorMsg>,
) {
'pump: loop {
tokio::select! {
Some(msg) = outbound_rx.next() => {
let msg = Frame::text(msg.content.into_bytes().into());
let _ = websocket.write_frame(msg).await;
}
result = websocket.read_frame() => {
match result {
Ok(msg) => match msg.opcode {
OpCode::Text => {
if let Ok(s) = String::from_utf8(msg.payload.to_vec()) {
let _ = inbound_tx.unbounded_send(s);
}
}
OpCode::Close => {
log::info!("Debugger session ended");
break 'pump;
}
_ => {
}
},
Err(_) => {
log::info!("Debugger session ended (connection lost)");
break 'pump;
}
}
}
else => {
break 'pump;
}
}
}
}
pub struct InspectorInfo {
pub host: SocketAddr,
pub uuid: Uuid,
pub thread_name: Option<String>,
pub new_session_tx: UnboundedSender<InspectorSessionProxy>,
pub deregister_rx: oneshot::Receiver<()>,
pub url: String,
pub wait_for_session: bool,
}
impl InspectorInfo {
pub fn new(
host: SocketAddr,
new_session_tx: mpsc::UnboundedSender<InspectorSessionProxy>,
deregister_rx: oneshot::Receiver<()>,
url: String,
wait_for_session: bool,
) -> Self {
Self {
host,
uuid: Uuid::new_v4(),
thread_name: thread::current().name().map(|n| n.to_owned()),
new_session_tx,
deregister_rx,
url,
wait_for_session,
}
}
fn get_json_metadata(&self, host: &Option<String>) -> Value {
let host_listen = format!("{}", self.host);
let host = host.as_ref().unwrap_or(&host_listen);
json!({
"description": "deno",
"devtoolsFrontendUrl": self.get_frontend_url(host),
"faviconUrl": "https://deno.land/favicon.ico",
"id": self.uuid.to_string(),
"title": self.get_title(),
"type": "node",
"url": self.url.to_string(),
"webSocketDebuggerUrl": self.get_websocket_debugger_url(host),
})
}
pub fn get_websocket_debugger_url(&self, host: &str) -> String {
format!("ws://{}/ws/{}", host, &self.uuid)
}
fn get_frontend_url(&self, host: &str) -> String {
format!(
"devtools://devtools/bundled/js_app.html?ws={}/ws/{}&experiments=true&v8only=true",
host, &self.uuid
)
}
fn get_title(&self) -> String {
format!(
"deno{} [pid: {}]",
self
.thread_name
.as_ref()
.map(|n| format!(" - {n}"))
.unwrap_or_default(),
process::id(),
)
}
}
pub struct MainInspectorSessionChannel(
Arc<Mutex<Option<UnboundedSender<InspectorSessionProxy>>>>,
);
impl MainInspectorSessionChannel {
pub fn new() -> Self {
Self(Arc::new(Mutex::new(None)))
}
pub fn set(&self, tx: UnboundedSender<InspectorSessionProxy>) {
*self.0.lock() = Some(tx);
}
pub fn get(&self) -> Option<UnboundedSender<InspectorSessionProxy>> {
self.0.lock().clone()
}
}
impl Default for MainInspectorSessionChannel {
fn default() -> Self {
Self::new()
}
}
impl Clone for MainInspectorSessionChannel {
fn clone(&self) -> Self {
Self(self.0.clone())
}
}