Repository: denoland/deno
Pull request: feat(unstable): notify control sock on serving

send an event over the control socket when deno.serve is listening. 

Changed files:
filename: cli/main.rs
patch: @@ -610,6 +610,8 @@ fn wait_for_start(
   Some(async move {
     use tokio::io::AsyncBufReadExt;
     use tokio::io::AsyncRead;
+    use tokio::io::AsyncWrite;
+    use tokio::io::AsyncWriteExt;
     use tokio::io::BufReader;
     use tokio::net::TcpListener;
     use tokio::net::UnixSocket;
@@ -630,18 +632,23 @@ fn wait_for_start(
       vec![],
     );
 
-    let stream: Box<dyn AsyncRead + Unpin> = match addr.split_once(':') {
+    let (rx, mut tx): (
+      Box<dyn AsyncRead + Unpin>,
+      Box<dyn AsyncWrite + Send + Unpin>,
+    ) = match addr.split_once(':') {
       Some(("tcp", addr)) => {
         let listener = TcpListener::bind(addr).await?;
         let (stream, _) = listener.accept().await?;
-        Box::new(stream)
+        let (rx, tx) = stream.into_split();
+        (Box::new(rx), Box::new(tx))
       }
       Some(("unix", path)) => {
         let socket = UnixSocket::new_stream()?;
         socket.bind(path)?;
         let listener = socket.listen(1)?;
         let (stream, _) = listener.accept().await?;
-        Box::new(stream)
+        let (rx, tx) = stream.into_split();
+        (Box::new(rx), Box::new(tx))
       }
       Some(("vsock", addr)) => {
         let Some((cid, port)) = addr.split_once(':') else {
@@ -652,17 +659,28 @@ fn wait_for_start(
         let addr = VsockAddr::new(cid, port);
         let listener = VsockListener::bind(addr)?;
         let (stream, _) = listener.accept().await?;
-        Box::new(stream)
+        let (rx, tx) = stream.into_split();
+        (Box::new(rx), Box::new(tx))
       }
       _ => {
         deno_core::anyhow::bail!("invalid control sock");
       }
     };
 
-    let mut stream = BufReader::new(stream);
-
     let mut buf = Vec::with_capacity(1024);
-    stream.read_until(b'\n', &mut buf).await?;
+    BufReader::new(rx).read_until(b'\n', &mut buf).await?;
+
+    tokio::spawn(async move {
+      deno_runtime::deno_http::SERVE_NOTIFIER.notified().await;
+
+      #[derive(deno_core::serde::Serialize)]
+      enum Event {
+        Serving,
+      }
+
+      let buf = deno_core::serde_json::to_vec(&Event::Serving).unwrap();
+      let _ = tx.write_all(&buf).await;
+    });
 
     #[derive(deno_core::serde::Deserialize)]
     struct Start {

filename: ext/http/00_serve.ts
patch: @@ -14,6 +14,7 @@ import {
   op_http_get_request_headers,
   op_http_get_request_method_and_url,
   op_http_metric_handle_otel_error,
+  op_http_notify_serving,
   op_http_read_request_body,
   op_http_request_on_cancel,
   op_http_serve,
@@ -1049,6 +1050,8 @@ function serveHttpOn(context, addr, callback) {
     }
   })();
 
+  op_http_notify_serving();
+
   return {
     addr,
     finished,

filename: ext/http/lib.rs
patch: @@ -15,6 +15,8 @@ use std::net::SocketAddr;
 use std::pin::pin;
 use std::pin::Pin;
 use std::rc::Rc;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::task::ready;
 use std::task::Context;
@@ -80,6 +82,7 @@ use serde::Serialize;
 use tokio::io::AsyncRead;
 use tokio::io::AsyncWrite;
 use tokio::io::AsyncWriteExt;
+use tokio::sync::Notify;
 
 use crate::network_buffered_stream::NetworkBufferedStream;
 use crate::reader_stream::ExternallyAbortableReaderStream;
@@ -201,6 +204,7 @@ deno_core::extension!(
     op_http_write_headers,
     op_http_write_resource,
     op_http_write,
+    op_http_notify_serving,
     http_next::op_http_close_after_finish,
     http_next::op_http_get_request_header,
     http_next::op_http_get_request_headers,
@@ -1764,6 +1768,20 @@ fn parse_serve_address(input: &str) -> (u8, String, u32, bool) {
   }
 }
 
+pub static SERVE_NOTIFIER: Notify = Notify::const_new();
+
+#[op2(fast)]
+fn op_http_notify_serving() {
+  static ONCE: AtomicBool = AtomicBool::new(false);
+
+  if ONCE
+    .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
+    .is_ok()
+  {
+    SERVE_NOTIFIER.notify_one();
+  }
+}
+
 #[cfg(test)]
 mod tests {
   use super::*;

filename: tests/specs/run/unconfigured/main.out
patch: @@ -1,3 +1,4 @@
 true
 hello world
+EVENT: "Serving"
 [Object: null prototype] { success: true, code: 0, signal: null }

filename: tests/specs/run/unconfigured/main.ts
patch: @@ -36,6 +36,7 @@ Deno.writeTextFile(
   `
 console.log(Deno[Deno.internal].isFromUnconfiguredRuntime());
 console.log(Deno.env.get('A'));
+Deno.serve({ onListen() {} }, () => {}).shutdown();
 `,
 );
 
@@ -47,4 +48,9 @@ const data = JSON.stringify({
 
 await sock.write(new TextEncoder().encode(data + "\n"));
 
+const buf = new Uint8Array(128);
+const n = await sock.read(buf);
+
+console.log("EVENT:", new TextDecoder().decode(buf.subarray(0, n)));
+
 console.log(await child.status);