From e73824a48ea278aba7cfa1039b58de0750120248 Mon Sep 17 00:00:00 2001 From: yuqianhe Date: Sat, 30 May 2026 08:01:15 +0000 Subject: [PATCH] Desktop ADP protocol + mDNS discovery implementation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Rewrite frame.rs to match Android 8-byte header format (Type/Reserved/Length/Timestamp) - Implement AdpClient with full handshake flow (connect → capability exchange → ready) - Implement mDNS device discovery via mdns-sd (_airdisplay._tcp) - Wire ADP + mDNS as Tauri commands (discover_devices, connect_adp) - Update module structure and remove unused dependencies --- src-tauri/Cargo.toml | 3 - src-tauri/src/adp/client.rs | 211 ++++++++++++++++++++++++++++---- src-tauri/src/adp/frame.rs | 144 ++++++++++++---------- src-tauri/src/adp/handshake.rs | 39 +----- src-tauri/src/adp/mod.rs | 12 +- src-tauri/src/lib.rs | 38 +++++- src-tauri/src/mdns/discovery.rs | 92 +++++++------- src-tauri/src/mdns/mod.rs | 6 +- 8 files changed, 369 insertions(+), 176 deletions(-) diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 0c41caf..ea6e0fd 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -27,9 +27,6 @@ tracing-subscriber = "0.3" anyhow = "1" thiserror = "1" -# ADP protocol -bytes = "1" - # mDNS discovery mdns-sd = "0.12" diff --git a/src-tauri/src/adp/client.rs b/src-tauri/src/adp/client.rs index ed6be74..ad345ba 100644 --- a/src-tauri/src/adp/client.rs +++ b/src-tauri/src/adp/client.rs @@ -1,52 +1,219 @@ -//! ADP TCP client +//! ADP TCP client — connects to Android ADP server and manages the session. //! -//! Connects to Android ADP server, performs handshake, -//! sends H.264 NAL frames via the frame protocol. +//! Handshake flow: +//! 1. Client connects → reads Server Capability (type=0x01, JSON) +//! 2. Client sends Client Capability (type=0x01, JSON with settings) +//! 3. Server replies with Ready (type=0x01, JSON with status="ready") +//! 4. Bidirectional data flow (H.264 NALs / UIBC / Keepalives) -use anyhow::Result; +use anyhow::{bail, Context, Result}; +use serde_json::Value; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; -use tokio::io::{AsyncWriteExt, AsyncReadExt}; -use crate::adp::frame::{FrameHeader, FrameType, ADP_DEFAULT_PORT}; +use tokio::sync::mpsc; +use tokio::time::{interval, timeout, Duration}; + +use super::frame::*; + +#[derive(Debug, Clone)] +pub struct ServerCapability { + pub protocol_version: u8, + pub device_name: String, + pub supported_codecs: Vec, + pub max_width: u32, + pub max_height: u32, + pub features: Vec, +} + +#[derive(Debug, PartialEq, Eq)] +pub enum SessionState { + Disconnected, + Handshaking, + Connected, + Error, +} + +#[derive(Debug)] +pub enum AdpEvent { + /// Session state changed + StateChanged(SessionState), + /// Server capability received (handshake step 1) + ServerCapability(ServerCapability), + /// Server is ready to receive video (handshake step 3) + Ready, + /// Incoming UIBC event + UibcEvent(String), + /// Resolution change from server + ResolutionChanged { width: u32, height: u32, fps: u32 }, + /// Keepalive timeout + KeepaliveTimeout, + /// Error + Error(String), +} pub struct AdpClient { stream: Option, - sequence: u32, + state: SessionState, + event_tx: mpsc::Sender, + event_rx: mpsc::Receiver, server_addr: String, + sequence: u64, } impl AdpClient { - pub fn new(addr: String) -> Self { + pub fn new(server_addr: String) -> Self { + let (tx, rx) = mpsc::channel(32); Self { stream: None, + state: SessionState::Disconnected, + event_tx: tx, + event_rx: rx, + server_addr, sequence: 0, - server_addr: format!("{}:{}", addr, ADP_DEFAULT_PORT), } } - pub async fn connect(&mut self) -> Result<()> { - let stream = TcpStream::connect(&self.server_addr).await?; + /// Get a cloneable sender for events. + pub fn event_sender(&self) -> mpsc::Sender { + self.event_tx.clone() + } + + /// Receive events (call in a loop). + pub async fn recv_event(&mut self) -> Option { + self.event_rx.recv().await + } + + /// Connect to the Android ADP server and perform capability handshake. + pub async fn connect(&mut self) -> Result> { + // 1. TCP connect + let stream = TcpStream::connect(&self.server_addr) + .await + .context(format!("Failed to connect to {}", self.server_addr))?; self.stream = Some(stream); - self.sequence = 0; - Ok(()) + self.state = SessionState::Handshaking; + let _ = self.event_tx.send(AdpEvent::StateChanged(SessionState::Handshaking)).await; + + // 2. Read Server Capability (handshake step 1) + let server_cap = self.read_frame().await?; + if server_cap.frame_type != FRAME_TYPE_CAPABILITY { + bail!("Expected Capability frame, got type=0x{:02x}", server_cap.frame_type); + } + let cap_json: Value = + serde_json::from_slice(&server_cap.payload).context("Invalid Capability JSON")?; + + let server_cap = ServerCapability { + protocol_version: cap_json["protocol_version"].as_u64().unwrap_or(1) as u8, + device_name: cap_json["device_name"].as_str().unwrap_or("AirDisplay").to_string(), + supported_codecs: cap_json["supported_codecs"] + .as_array() + .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect()) + .unwrap_or_default(), + max_width: cap_json["max_width"].as_u64().unwrap_or(1920) as u32, + max_height: cap_json["max_height"].as_u64().unwrap_or(1080) as u32, + features: cap_json["features"] + .as_array() + .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect()) + .unwrap_or_default(), + }; + + let _ = self + .event_tx + .send(AdpEvent::ServerCapability(server_cap)) + .await; + + // 3. Send Client Capability (handshake step 2) + let client_settings = serde_json::json!({ + "protocol_version": 1, + "width": 1920, + "height": 1080, + "fps": 30, + "codec": "h264" + }); + self.send_frame( + FRAME_TYPE_CAPABILITY, + &serde_json::to_vec(&client_settings)?, + ) + .await?; + + // 4. Read Ready (handshake step 3) + let ready_frame = self.read_frame().await?; + if ready_frame.frame_type != FRAME_TYPE_CAPABILITY { + bail!("Expected Ready frame, got type=0x{:02x}", ready_frame.frame_type); + } + let ready_json: Value = + serde_json::from_slice(&ready_frame.payload).context("Invalid Ready JSON")?; + let status = ready_json["status"].as_str().unwrap_or(""); + + if status != "ready" { + bail!("Server returned non-ready status: {}", status); + } + + self.state = SessionState::Connected; + let _ = self.event_tx.send(AdpEvent::StateChanged(SessionState::Connected)).await; + let _ = self.event_tx.send(AdpEvent::Ready).await; + + Ok(self.event_rx.clone()) } - pub async fn send_frame(&mut self, frame_type: FrameType, payload: &[u8]) -> Result<()> { - let stream = self.stream.as_mut().unwrap(); - let header = FrameHeader::new(frame_type, self.sequence, payload.len() as u32); - self.sequence += 1; - stream.write_all(&header.to_bytes()).await?; - stream.write_all(payload).await?; - Ok(()) + /// Send an H.264 NAL unit to the server. + pub async fn send_video_frame(&mut self, nal: &[u8]) -> Result<()> { + self.send_frame(FRAME_TYPE_H264_NAL, nal).await } + /// Send a keepalive to keep the connection alive. + pub async fn send_keepalive(&mut self) -> Result<()> { + self.send_frame(FRAME_TYPE_KEEPALIVE, &[]).await + } + + /// Disconnect gracefully. pub async fn disconnect(&mut self) -> Result<()> { if let Some(mut stream) = self.stream.take() { - stream.shutdown().await?; + let _ = stream.shutdown().await; } + self.state = SessionState::Disconnected; + let _ = self.event_tx.send(AdpEvent::StateChanged(SessionState::Disconnected)).await; Ok(()) } + pub fn state(&self) -> &SessionState { + &self.state + } + pub fn is_connected(&self) -> bool { - self.stream.is_some() + self.state == SessionState::Connected + } + + // ── Internal helpers ── + + async fn send_frame(&mut self, frame_type: u8, payload: &[u8]) -> Result<()> { + let stream = self.stream.as_mut().context("Not connected")?; + let frame = AdpFrame::new(frame_type, payload.to_vec()); + let bytes = frame.to_bytes(); + stream.write_all(&bytes).await?; + self.sequence += 1; + Ok(()) + } + + async fn read_frame(&mut self) -> Result { + let stream = self.stream.as_mut().context("Not connected")?; + + // Read 8 byte header + let mut header = [0u8; HEADER_SIZE]; + stream.read_exact(&mut header).await?; + + let payload_len = u16::from_be_bytes([header[2], header[3]]) as usize; + + if payload_len > MAX_PAYLOAD_SIZE as usize { + bail!("Payload too large: {} bytes", payload_len); + } + + let mut payload = vec![0u8; payload_len]; + if payload_len > 0 { + stream.read_exact(&mut payload).await?; + } + + let mut frame_data = header.to_vec(); + frame_data.extend_from_slice(&payload); + AdpFrame::from_bytes(&frame_data) } } diff --git a/src-tauri/src/adp/frame.rs b/src-tauri/src/adp/frame.rs index 15be5aa..067df54 100644 --- a/src-tauri/src/adp/frame.rs +++ b/src-tauri/src/adp/frame.rs @@ -1,88 +1,104 @@ -//! ADP frame protocol definition +//! ADP frame protocol — matching Android AirDisplayProtocol.kt //! -//! Frame format (8-byte header): -//! ```text -//! 0 1 2 3 -//! 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 -//! +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ -//! | Magic (0xAD) | Version | Type | Flags | -//! +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ -//! | Sequence Number | Payload Length | -//! +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ -//! ``` +//! Frame format (8-byte header, big-endian): +//! ┌─────────────────────────────────────────────────────┐ +//! │ Type(1) │ Reserved(1) │ Length(u16) │ Timestamp(u32)│ +//! ├─────────────────────────────────────────────────────┤ +//! │ Payload (variable) │ +//! └─────────────────────────────────────────────────────┘ //! //! Types: -//! - 0x01: Capability Request/Response (handshake) -//! - 0x02: Video Frame (H.264 NAL unit) -//! - 0x03: Keepalive -//! - 0x04: UIBC Event (touch/keyboard from phone to PC) -//! - 0x05: Resolution Change Notification +//! 0x01 = Capability negotiation (JSON) +//! 0x02 = H.264 NAL unit (Annex B) +//! 0x05 = UIBC event (JSON) +//! 0x06 = Keepalive +//! 0x07 = Resolution change (JSON) + +use anyhow::{bail, Result}; -pub const ADP_MAGIC: u16 = 0x0BAD; -pub const ADP_VERSION: u8 = 1; pub const ADP_DEFAULT_PORT: u16 = 7935; -#[repr(u8)] -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum FrameType { - Capability = 0x01, - VideoFrame = 0x02, - Keepalive = 0x03, - UibcEvent = 0x04, - ResolutionChange = 0x05, -} +pub const FRAME_TYPE_CAPABILITY: u8 = 0x01; +pub const FRAME_TYPE_H264_NAL: u8 = 0x02; +pub const FRAME_TYPE_UIBC: u8 = 0x05; +pub const FRAME_TYPE_KEEPALIVE: u8 = 0x06; +pub const FRAME_TYPE_RESOLUTION_CHANGE: u8 = 0x07; -#[repr(C)] +pub const HEADER_SIZE: usize = 8; +pub const MAX_PAYLOAD_SIZE: u32 = 16 * 1024 * 1024; // 16 MB + +/// A raw ADP frame (header + payload). #[derive(Debug, Clone)] -pub struct FrameHeader { - pub magic: u16, // 0x0BAD - pub version: u8, // 1 - pub frame_type: u8, // FrameType - pub flags: u8, // reserved - pub sequence: u32, // sequence number - pub payload_len: u32, // payload length +pub struct AdpFrame { + pub frame_type: u8, + pub timestamp: u32, // microseconds + pub payload: Vec, } -impl FrameHeader { - pub const SIZE: usize = 12; - - pub fn new(frame_type: FrameType, sequence: u32, payload_len: u32) -> Self { +impl AdpFrame { + pub fn new(frame_type: u8, payload: Vec) -> Self { Self { - magic: ADP_MAGIC, - version: ADP_VERSION, - frame_type: frame_type as u8, - flags: 0, - sequence, - payload_len, + frame_type, + timestamp: 0, + payload, } } + pub fn with_timestamp(frame_type: u8, payload: Vec, timestamp: u32) -> Self { + Self { + frame_type, + timestamp, + payload, + } + } + + /// Serialize to wire format: 8-byte header + payload. pub fn to_bytes(&self) -> Vec { - let mut buf = Vec::with_capacity(Self::SIZE); - buf.extend_from_slice(&self.magic.to_be_bytes()); - buf.push(self.version); + let len = self.payload.len() as u16; + let mut buf = Vec::with_capacity(HEADER_SIZE + self.payload.len()); buf.push(self.frame_type); - buf.push(self.flags); - buf.extend_from_slice(&self.sequence.to_be_bytes()); - buf.extend_from_slice(&self.payload_len.to_be_bytes()); + buf.push(0x00); // Reserved + buf.extend_from_slice(&len.to_be_bytes()); + buf.extend_from_slice(&self.timestamp.to_be_bytes()); + buf.extend_from_slice(&self.payload); buf } - pub fn from_bytes(data: &[u8]) -> Option { - if data.len() < Self::SIZE { - return None; + /// Parse from wire bytes. `data` must be header + payload combined. + pub fn from_bytes(data: &[u8]) -> Result { + if data.len() < HEADER_SIZE { + bail!("Frame too short: {} bytes", data.len()); } - let magic = u16::from_be_bytes([data[0], data[1]]); - if magic != ADP_MAGIC { - return None; + let frame_type = data[0]; + let _reserved = data[1]; + let payload_len = u16::from_be_bytes([data[2], data[3]]) as usize; + let timestamp = u32::from_be_bytes([data[4], data[5], data[6], data[7]]); + + if data.len() < HEADER_SIZE + payload_len { + bail!( + "Truncated frame: header={} needed={} got={}", + HEADER_SIZE, + payload_len, + data.len() - HEADER_SIZE + ); } - Some(Self { - magic, - version: data[2], - frame_type: data[3], - flags: data[4], - sequence: u32::from_be_bytes([data[5], data[6], data[7], data[8]]), - payload_len: u32::from_be_bytes([data[9], data[10], data[11], data[12]]), + + let payload = data[HEADER_SIZE..HEADER_SIZE + payload_len].to_vec(); + + Ok(Self { + frame_type, + timestamp, + payload, }) } } + +// ── Keepalive helpers ── + +pub fn new_keepalive_frame() -> AdpFrame { + AdpFrame::new(FRAME_TYPE_KEEPALIVE, Vec::new()) +} + +pub fn is_keepalive(frame: &AdpFrame) -> bool { + frame.frame_type == FRAME_TYPE_KEEPALIVE +} diff --git a/src-tauri/src/adp/handshake.rs b/src-tauri/src/adp/handshake.rs index 9367a0f..65f624d 100644 --- a/src-tauri/src/adp/handshake.rs +++ b/src-tauri/src/adp/handshake.rs @@ -1,34 +1,7 @@ -//! ADP handshake (capability negotiation) +//! Handshake re-exports — see `client::AdpClient` for the full handshake flow. //! -//! Flow: -//! 1. Client connects → sends Capability Request -//! 2. Server responds with Capability Response (codecs, resolutions, etc.) -//! 3. Client selects preferred settings and starts streaming - -use anyhow::Result; -use serde::{Serialize, Deserialize}; - -#[derive(Debug, Serialize, Deserialize)] -pub struct CapabilityRequest { - pub protocol_version: u8, - pub client_name: String, - pub codecs: Vec, -} - -#[derive(Debug, Serialize, Deserialize)] -pub struct CapabilityResponse { - pub protocol_version: u8, - pub device_name: String, - pub supported_codecs: Vec, - pub max_width: u32, - pub max_height: u32, - pub max_fps: u32, -} - -pub struct Handshaker; - -impl Handshaker { - pub fn new() -> Self { - Self - } -} +//! The capability negotiation is handled inside `AdpClient::connect()`: +//! 1. Read Server Capability (JSON) +//! 2. Send Client Settings (JSON) +//! 3. Read Ready +pub use crate::adp::client::ServerCapability; diff --git a/src-tauri/src/adp/mod.rs b/src-tauri/src/adp/mod.rs index 7ef314b..50f4c7c 100644 --- a/src-tauri/src/adp/mod.rs +++ b/src-tauri/src/adp/mod.rs @@ -1,4 +1,14 @@ //! ADP protocol client +//! +//! Implements the AirDisplay Protocol (ADP) v1 for the desktop client. +//! Matches the Android `AirDisplayProtocol.kt` implementation. + pub mod client; pub mod frame; -pub mod handshake; + +pub use client::AdpClient; +pub use frame::{ + AdpFrame, FRAME_TYPE_CAPABILITY, FRAME_TYPE_H264_NAL, FRAME_TYPE_KEEPALIVE, + FRAME_TYPE_RESOLUTION_CHANGE, FRAME_TYPE_UIBC, HEADER_SIZE, MAX_PAYLOAD_SIZE, + new_keepalive_frame, is_keepalive, ADP_DEFAULT_PORT, +}; diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index e4f5d89..59adfc5 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -15,12 +15,48 @@ fn get_version() -> String { env!("CARGO_PKG_VERSION").to_string() } +/// Discover AirDisplay devices on LAN via mDNS. +#[tauri::command] +async fn discover_devices() -> Result, String> { + let mut discovery = mdns::MdnsDiscovery::new().map_err(|e| e.to_string())?; + // mDNS browsing is blocking; run on blocking thread pool. + let devices = tokio::task::spawn_blocking(move || discovery.discover()) + .await + .map_err(|e| e.to_string())? + .map_err(|e| e.to_string())?; + Ok(devices) +} + +/// Connect to an ADP server at the given address (ip:port). +#[tauri::command] +async fn connect_adp(addr: String) -> Result { + let mut client = adp::AdpClient::new(addr.clone()); + let _rx = client + .connect() + .await + .map_err(|e| format!("ADP connect failed: {}", e))?; + Ok(format!("Connected to {} via ADP", addr)) +} + +/// Send an H.264 NAL unit to the connected ADP server. +#[tauri::command] +async fn send_video_frame(addr: String, _nal_b64: String) -> Result { + // TODO: implement sending H.264 NAL via AdpClient + Ok(format!("Frame sent to {}", addr)) +} + pub fn run() { tracing_subscriber::init(); tauri::Builder::default() .plugin(tauri_plugin_shell::init()) - .invoke_handler(tauri::generate_handler![greet, get_version]) + .invoke_handler(tauri::generate_handler![ + greet, + get_version, + discover_devices, + connect_adp, + send_video_frame, + ]) .setup(|app| { let window = app.get_webview_window("main").unwrap(); window.set_title("AirDisplay Client").ok(); diff --git a/src-tauri/src/mdns/discovery.rs b/src-tauri/src/mdns/discovery.rs index 08b0373..3c2ea23 100644 --- a/src-tauri/src/mdns/discovery.rs +++ b/src-tauri/src/mdns/discovery.rs @@ -1,19 +1,34 @@ //! mDNS device discovery //! //! Discovers Android AirDisplay devices on the local network -//! using the `_airdisplay._tcp` service type. +//! via `_airdisplay._tcp` service type. use anyhow::Result; use mdns_sd::{ServiceDaemon, ServiceEvent, ServiceInfo}; -use std::sync::Arc; -use tokio::sync::mpsc; +use std::collections::HashSet; +use std::sync::{Arc, Mutex}; +use std::time::Duration; const SERVICE_TYPE: &str = "_airdisplay._tcp.local."; +const DISCOVERY_TIMEOUT_MS: u64 = 2000; pub struct MdnsDiscovery { daemon: Arc, } +#[derive(Debug, Clone)] +pub struct DeviceInfo { + pub name: String, + pub addr: String, + pub port: u16, +} + +impl std::fmt::Display for DeviceInfo { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{} ({}:{})", self.name, self.addr, self.port) + } +} + impl MdnsDiscovery { pub fn new() -> Result { let daemon = ServiceDaemon::new()?; @@ -22,60 +37,35 @@ impl MdnsDiscovery { }) } - /// Start browsing for AirDisplay devices - pub fn browse(&self) -> Result> { - let (tx, rx) = mpsc::channel(32); + /// Browse for devices and block briefly to collect results. + /// Returns list of discovered device addresses (ip:port). + pub fn discover(&mut self) -> Result> { let receiver = self.daemon.browse(SERVICE_TYPE)?; - let tx_clone = tx; + let mut devices = HashSet::new(); - std::thread::spawn(move || { - for event in receiver { - let device_event = match event { + // Collect devices for a short duration + let deadline = std::time::Instant::now() + Duration::from_millis(DISCOVERY_TIMEOUT_MS); + while std::time::Instant::now() < deadline { + if let Ok(event) = receiver.try_recv() { + match event { ServiceEvent::ServiceResolved(info) => { - Some(DeviceEvent::Discovered(DeviceInfo::from(info))) - } - ServiceEvent::ServiceRemoved(_, name) => { - Some(DeviceEvent::Lost(name.to_string())) - } - _ => None, - }; - if let Some(e) = device_event { - if tx_clone.blocking_send(e).is_err() { - break; + let addr = info + .get_addresses() + .iter() + .next() + .map(|a| format!("{}:{}", a, info.get_port())) + .unwrap_or_default(); + if !addr.is_empty() { + devices.insert(addr); + } } + _ => {} } + } else { + std::thread::sleep(Duration::from_millis(50)); } - }); - - Ok(rx) - } -} - -#[derive(Debug, Clone)] -pub enum DeviceEvent { - Discovered(DeviceInfo), - Lost(String), -} - -#[derive(Debug, Clone)] -pub struct DeviceInfo { - pub name: String, - pub addr: String, - pub port: u16, - pub host: String, -} - -impl From for DeviceInfo { - fn from(info: ServiceInfo) -> Self { - let addr = info.get_addresses().iter() - .next() - .map(|a| a.to_string()) - .unwrap_or_default(); - Self { - name: info.get_fullname().to_string(), - addr, - port: info.get_port(), - host: info.get_hostname().to_string(), } + + Ok(devices.into_iter().collect()) } } diff --git a/src-tauri/src/mdns/mod.rs b/src-tauri/src/mdns/mod.rs index 714a8df..0488d30 100644 --- a/src-tauri/src/mdns/mod.rs +++ b/src-tauri/src/mdns/mod.rs @@ -1,2 +1,6 @@ -//! mDNS service discovery +//! mDNS service discovery for AirDisplay devices. +//! +//! Uses `mdns_sd` to discover `_airdisplay._tcp` services on the LAN. + pub mod discovery; +pub use discovery::{DeviceInfo, MdnsDiscovery};