fix: Add mobile mDNS service metadata handling

qdot · qdot · commit acc9924ce9bb · 2026-05-31T16:07:59.000-07:00
diff --git a/crates/buttplug_transport_websocket_tungstenite/src/websocket_server.rs b/crates/buttplug_transport_websocket_tungstenite/src/websocket_server.rs
@@ -16,7 +16,7 @@ use buttplug_core::{
   message::serializer::ButtplugSerializedMessage,
 };
 use futures::{FutureExt, SinkExt, StreamExt, future::BoxFuture};
-use std::{sync::Arc, time::Duration};
+use std::{fmt, sync::Arc, time::Duration};
 use tokio::{
   net::{TcpListener, TcpStream},
   select,
@@ -27,19 +27,42 @@ use tokio::{
   time::sleep,
 };
 
+#[derive(Clone)]
+struct ListenerBoundCallback(Arc<dyn Fn(u16) + Send + Sync>);
+
+impl ListenerBoundCallback {
+  fn new(callback: impl Fn(u16) + Send + Sync + 'static) -> Self {
+    Self(Arc::new(callback))
+  }
+
+  fn call(&self, port: u16) {
+    (self.0)(port);
+  }
+}
+
+impl fmt::Debug for ListenerBoundCallback {
+  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+    f.debug_struct("ListenerBoundCallback")
+      .finish_non_exhaustive()
+  }
+}
+
 #[derive(Clone, Debug)]
 pub struct ButtplugWebsocketServerTransportBuilder {
   /// If true, listens all on available interfaces. Otherwise, only listens on 127.0.0.1.
   listen_on_all_interfaces: bool,
   /// Insecure port for listening for websocket connections.
   port: u16,
+  /// Optional callback fired after the listener is bound and the actual local port is known.
+  listener_bound_callback: Option<ListenerBoundCallback>,
 }
 
 impl Default for ButtplugWebsocketServerTransportBuilder {
   fn default() -> Self {
     Self {
       listen_on_all_interfaces: false,
       port: 12345,
+      listener_bound_callback: None,
     }
   }
 }
@@ -55,10 +78,16 @@ impl ButtplugWebsocketServerTransportBuilder {
     self
   }
 
+  pub fn on_listener_bound(&mut self, callback: impl Fn(u16) + Send + Sync + 'static) -> &mut Self {
+    self.listener_bound_callback = Some(ListenerBoundCallback::new(callback));
+    self
+  }
+
   pub fn finish(&self) -> ButtplugWebsocketServerTransport {
     ButtplugWebsocketServerTransport {
       port: self.port,
       listen_on_all_interfaces: self.listen_on_all_interfaces,
+      listener_bound_callback: self.listener_bound_callback.clone(),
       disconnect_notifier: Arc::new(Notify::new()),
     }
   }
@@ -193,6 +222,7 @@ async fn run_connection_loop(
 pub struct ButtplugWebsocketServerTransport {
   port: u16,
   listen_on_all_interfaces: bool,
+  listener_bound_callback: Option<ListenerBoundCallback>,
   disconnect_notifier: Arc<Notify>,
 }
 
@@ -203,6 +233,7 @@ impl ButtplugConnectorTransport for ButtplugWebsocketServerTransport {
     incoming_sender: Sender<ButtplugTransportIncomingMessage>,
   ) -> BoxFuture<'static, Result<(), ButtplugConnectorError>> {
     let disconnect_notifier = self.disconnect_notifier.clone();
+    let listener_bound_callback = self.listener_bound_callback.clone();
 
     let base_addr = if self.listen_on_all_interfaces {
       "0.0.0.0"
@@ -231,6 +262,19 @@ impl ButtplugConnectorTransport for ButtplugWebsocketServerTransport {
         )
       })?;
       debug!("Websocket: Listening on: {}", addr);
+      if let Some(callback) = &listener_bound_callback {
+        let local_port = listener
+          .local_addr()
+          .map_err(|e| {
+            ButtplugConnectorError::TransportSpecificError(
+              ButtplugConnectorTransportSpecificError::GenericNetworkError(format!(
+                "Could not determine websocket listener local address: {e}"
+              )),
+            )
+          })?
+          .port();
+        callback.call(local_port);
+      }
       if let Ok((stream, _)) = listener.accept().await {
         info!("Websocket: Got connection");
         let ws_stream = tokio_tungstenite::accept_async(stream)
@@ -288,6 +332,7 @@ mod test {
     message::serializer::ButtplugSerializedMessage,
   };
   use std::io::ErrorKind;
+  use std::sync::{Arc, Mutex};
   use tokio::{net::TcpListener, sync::mpsc};
 
   #[tokio::test]
@@ -322,4 +367,37 @@ mod test {
       other => panic!("Unexpected error: {other:?}"),
     }
   }
+
+  #[tokio::test]
+  async fn listener_bound_callback_receives_actual_port() {
+    let bound_port = Arc::new(Mutex::new(None));
+    let callback_port = bound_port.clone();
+    let transport = ButtplugWebsocketServerTransportBuilder::default()
+      .on_listener_bound(move |port| {
+        *callback_port.lock().unwrap() = Some(port);
+      })
+      .finish();
+    let (_outgoing_sender, outgoing_receiver) = mpsc::channel::<ButtplugSerializedMessage>(1);
+    let (incoming_sender, _incoming_receiver) =
+      mpsc::channel::<ButtplugTransportIncomingMessage>(1);
+    let connect_task = tokio::spawn(async move {
+      let _ = transport.connect(outgoing_receiver, incoming_sender).await;
+    });
+
+    tokio::time::timeout(std::time::Duration::from_secs(1), async {
+      loop {
+        if let Some(port) = *bound_port.lock().unwrap() {
+          return port;
+        }
+        tokio::task::yield_now().await;
+      }
+    })
+    .await
+    .expect("listener bound callback was not called");
+
+    let port = bound_port.lock().unwrap().unwrap();
+    assert!(port > 0);
+
+    connect_task.abort();
+  }
 }
diff --git a/crates/intiface_engine/src/buttplug_server.rs b/crates/intiface_engine/src/buttplug_server.rs
@@ -164,18 +164,23 @@ pub async fn setup_buttplug_server(
 pub async fn run_server(
   server: &ButtplugRemoteServer,
   options: &EngineOptions,
+  on_listener_bound: Option<Arc<dyn Fn(u16) + Send + Sync>>,
 ) -> Result<(), ButtplugServerConnectorError> {
   if let Some(port) = options.websocket_port() {
+    let mut transport_builder = ButtplugWebsocketServerTransportBuilder::default();
+    transport_builder
+      .port(port)
+      .listen_on_all_interfaces(options.websocket_use_all_interfaces());
+    if let Some(on_listener_bound) = on_listener_bound {
+      transport_builder.on_listener_bound(move |bound_port| {
+        on_listener_bound(bound_port);
+      });
+    }
     server
       .start(ButtplugRemoteServerConnector::<
         _,
         ButtplugServerJSONSerializer,
-      >::new(
-        ButtplugWebsocketServerTransportBuilder::default()
-          .port(port)
-          .listen_on_all_interfaces(options.websocket_use_all_interfaces())
-          .finish(),
-      ))
+      >::new(transport_builder.finish()))
       .await
   } else if let Some(addr) = options.websocket_client_address() {
     server
diff --git a/crates/intiface_engine/src/engine.rs b/crates/intiface_engine/src/engine.rs
@@ -14,7 +14,7 @@ use crate::{
     Frontend, frontend_external_event_loop, frontend_server_event_loop,
     process_messages::{EngineErrorDetail, EngineMessage},
   },
-  mdns::IntifaceMdns,
+  mdns::{IntifaceMdns, IntifaceMdnsServiceMetadata},
   options::EngineOptions,
   remote_server::{ButtplugRemoteServerEvent, ButtplugServerConnectorError},
   rest_server::IntifaceRestServer,
@@ -26,7 +26,15 @@ use buttplug_core::connector::{
 use buttplug_server_device_config::{DeviceConfigurationManager, save_user_config};
 use futures::{StreamExt, pin_mut};
 use once_cell::sync::OnceCell;
-use std::{io::ErrorKind, path::Path, sync::Arc, time::Duration};
+use std::{
+  io::ErrorKind,
+  path::Path,
+  sync::{
+    Arc, Mutex,
+    atomic::{AtomicBool, Ordering},
+  },
+  time::Duration,
+};
 use tokio::{fs, select};
 use tokio_util::sync::CancellationToken;
 
@@ -90,6 +98,39 @@ fn websocket_port_in_use_error(err: &ButtplugServerConnectorError) -> Option<(St
   }
 }
 
+fn engine_server_created_message(
+  metadata: Option<&IntifaceMdnsServiceMetadata>,
+  port: Option<u16>,
+) -> EngineMessage {
+  if let Some(metadata) = metadata {
+    EngineMessage::EngineServerCreated {
+      service_type: Some(metadata.service_type.clone()),
+      instance_name: Some(metadata.instance_name.clone()),
+      port,
+      txt_records: Some(metadata.txt_records.clone()),
+    }
+  } else {
+    EngineMessage::EngineServerCreated {
+      service_type: None,
+      instance_name: None,
+      port: None,
+      txt_records: None,
+    }
+  }
+}
+
+async fn send_engine_server_created(
+  frontend: Option<&Arc<dyn Frontend>>,
+  metadata: Option<&IntifaceMdnsServiceMetadata>,
+  port: Option<u16>,
+) {
+  if let Some(frontend) = frontend {
+    frontend
+      .send(engine_server_created_message(metadata, port))
+      .await;
+  }
+}
+
 impl IntifaceEngine {
   pub fn backdoor_server(&self) -> Option<Arc<BackdoorServer>> {
     Some(self.backdoor_server.get()?.clone())
@@ -112,15 +153,41 @@ impl IntifaceEngine {
       frontend.send(EngineMessage::EngineStarted {}).await;
     }
 
-    // Set up mDNS
-    let _mdns_server = if options.broadcast_server_mdns() {
-      // TODO Unregister whenever we have a live connection
-
-      // TODO Support different services for engine versus repeater
-      IntifaceMdns::new()
-    } else {
-      None
-    };
+    let mdns_service_metadata =
+      if options.broadcast_server_mdns() && options.websocket_port().is_some() {
+        Some(Arc::new(IntifaceMdnsServiceMetadata::new(
+          options.mdns_suffix().as_deref(),
+        )))
+      } else {
+        None
+      };
+    let mdns_publisher = Arc::new(Mutex::new(None::<IntifaceMdns>));
+    let engine_server_created_sent = Arc::new(AtomicBool::new(false));
+    let on_listener_bound = mdns_service_metadata.as_ref().map(|metadata| {
+      let metadata = Arc::clone(metadata);
+      let frontend = frontend.clone();
+      let mdns_publisher = Arc::clone(&mdns_publisher);
+      let engine_server_created_sent = Arc::clone(&engine_server_created_sent);
+      Arc::new(move |port: u16| {
+        if !engine_server_created_sent.swap(true, Ordering::SeqCst) {
+          let metadata = Arc::clone(&metadata);
+          let frontend = frontend.clone();
+          tokio::spawn(async move {
+            send_engine_server_created(frontend.as_ref(), Some(metadata.as_ref()), Some(port))
+              .await;
+          });
+        }
+        #[cfg(target_os = "ios")]
+        info!("Skipping Rust mDNS publisher on iOS; host app is expected to publish Bonjour");
+        #[cfg(not(target_os = "ios"))]
+        {
+          let mut publisher = mdns_publisher.lock().unwrap();
+          if publisher.is_none() {
+            *publisher = IntifaceMdns::new(metadata.as_ref(), port);
+          }
+        }
+      }) as Arc<dyn Fn(u16) + Send + Sync>
+    });
 
     // Set up Repeater (if in repeater mode)
     if options.repeater_mode() {
@@ -233,7 +300,9 @@ impl IntifaceEngine {
       }
     }
     if let Some(frontend) = &frontend {
-      frontend.send(EngineMessage::EngineServerCreated {}).await;
+      if on_listener_bound.is_none() {
+        send_engine_server_created(Some(frontend), None, None).await;
+      }
       let event_receiver = server.event_stream();
       let frontend_clone = frontend.clone();
       let stop_child_token = self.stop_token.child_token();
@@ -257,7 +326,7 @@ impl IntifaceEngine {
           info!("Owner requested process exit, exiting.");
           exit_requested = true;
         }
-        result = run_server(&server, options) => {
+        result = run_server(&server, options, on_listener_bound.clone()) => {
           match result {
             Ok(_) => info!("Connection dropped, restarting stay open loop."),
             Err(e) => {
diff --git a/crates/intiface_engine/src/frontend/process_messages.rs b/crates/intiface_engine/src/frontend/process_messages.rs
@@ -27,7 +27,16 @@ pub enum EngineMessage {
     #[serde(skip_serializing_if = "Option::is_none")]
     detail: Option<EngineErrorDetail>,
   },
-  EngineServerCreated {},
+  EngineServerCreated {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    service_type: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    instance_name: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    port: Option<u16>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    txt_records: Option<Vec<String>>,
+  },
   EngineStopped {},
   ClientConnected {
     client_name: String,
@@ -103,4 +112,42 @@ mod test {
       }),
     );
   }
+
+  #[test]
+  fn engine_server_created_serializes_optional_service_metadata() {
+    let message = EngineMessage::EngineServerCreated {
+      service_type: Some("_intiface_engine._tcp".to_owned()),
+      instance_name: Some("Intiface ABC123".to_owned()),
+      port: Some(12345),
+      txt_records: Some(vec!["path=/".to_owned()]),
+    };
+
+    assert_eq!(
+      serde_json::to_value(message).unwrap(),
+      json!({
+        "EngineServerCreated": {
+          "service_type": "_intiface_engine._tcp",
+          "instance_name": "Intiface ABC123",
+          "port": 12345,
+          "txt_records": ["path=/"]
+        }
+      }),
+    );
+  }
+
+  #[test]
+  fn engine_server_created_deserializes_legacy_empty_payload() {
+    let message: EngineMessage =
+      serde_json::from_value(json!({"EngineServerCreated": {}})).unwrap();
+
+    assert!(matches!(
+      message,
+      EngineMessage::EngineServerCreated {
+        service_type: None,
+        instance_name: None,
+        port: None,
+        txt_records: None,
+      }
+    ));
+  }
 }
diff --git a/crates/intiface_engine/src/mdns.rs b/crates/intiface_engine/src/mdns.rs
