Merge pull request #436 from lichon/main

add webrtc stream

Author: rustdesk

Cargo.toml

@@ -6,6 +6,10 @@ edition = "2018"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
+[features]
+default = []
+webrtc = ["dep:webrtc"]
+
 [dependencies]
 # new flexi_logger failed on rustc 1.75
 flexi_logger = { version = "0.27", features = ["async"] }
@@ -61,6 +65,7 @@ rustls-pki-types = "1.11"
 rustls-native-certs = "0.8"
 webpki-roots = "1.0.4"
 async-recursion = "1.1"
+webrtc = { version = "0.14.0", optional = true }
 
 [target.'cfg(not(any(target_os = "android", target_os = "ios")))'.dependencies]
 mac_address = "1.1"
@@ -70,6 +75,10 @@ machine-uid = { git = "https://github.com/rustdesk-org/machine-uid" }
 [build-dependencies]
 protobuf-codegen = { version = "3.7" }
 
+[dev-dependencies]
+clap = "4.5.51"
+webrtc = "0.14.0"
+
 [target.'cfg(target_os = "windows")'.dependencies]
 winapi = { version = "0.3", features = [
     "winuser",

examples/webrtc.rs

@@ -0,0 +1,154 @@
+extern crate hbb_common;
+
+#[cfg(feature = "webrtc")]
+use hbb_common::webrtc::WebRTCStream;
+
+use std::io::Write;
+use anyhow::Result;
+use bytes::Bytes;
+use clap::{Arg, Command};
+use tokio::time::Duration;
+
+#[cfg(not(feature = "webrtc"))]
+#[tokio::main]
+async fn main() -> Result<()> {
+    println!(
+        "The webrtc feature is not enabled. \
+        Please enable the webrtc feature to run this example."
+    );
+    Ok(())
+}
+
+#[cfg(feature = "webrtc")]
+#[tokio::main]
+async fn main() -> Result<()> {
+    let app = Command::new("webrtc-stream")
+        .about("An example of webrtc stream using hbb_common and webrtc-rs")
+        .arg(
+            Arg::new("debug")
+                .long("debug")
+                .short('d')
+                .action(clap::ArgAction::SetTrue)
+                .help("Prints debug log information"),
+        )
+        .arg(
+            Arg::new("offer")
+                .long("offer")
+                .short('o')
+                .help("set offer from other endpoint"),
+        );
+
+    let matches = app.clone().get_matches();
+
+    let debug = matches.contains_id("debug");
+    if debug {
+        println!("Debug log enabled");
+        env_logger::Builder::new()
+            .format(|buf, record| {
+                writeln!(
+                    buf,
+                    "{}:{} [{}] {} - {}",
+                    record.file().unwrap_or("unknown"),
+                    record.line().unwrap_or(0),
+                    record.level(),
+                    chrono::Local::now().format("%H:%M:%S.%6f"),
+                    record.args()
+                )
+            })
+            .filter(Some("hbb_common"), log::LevelFilter::Debug)
+            .init();
+    }
+
+    let remote_endpoint = if let Some(endpoint) = matches.get_one::<String>("offer") {
+        endpoint.to_string()
+    } else {
+        "".to_string()
+    };
+
+    let webrtc_stream = WebRTCStream::new(&remote_endpoint, false, 30000).await?;
+    // Print the offer to be sent to the other peer
+    let local_endpoint = webrtc_stream.get_local_endpoint().await?;
+
+    if remote_endpoint.is_empty() {
+        println!();
+        // Wait for the answer to be pasted
+        println!(
+            "Start new terminal run: \n{} \ncopy remote endpoint and paste here",
+            format!(
+                "cargo r --features webrtc --example webrtc -- --offer {}",
+                local_endpoint
+            )
+        );
+        // readline blocking
+        let line = std::io::stdin()
+            .lines()
+            .next()
+            .ok_or_else(|| anyhow::anyhow!("No input received"))??;
+        webrtc_stream.set_remote_endpoint(&line).await?;
+    } else {
+        println!(
+            "Copy local endpoint and paste to the other peer: \n{}",
+            local_endpoint
+        );
+    }
+
+    let s1 = webrtc_stream.clone();
+    tokio::spawn(async move {
+        let _ = read_loop(s1).await;
+    });
+
+    let s2 = webrtc_stream.clone();
+    tokio::spawn(async move {
+        let _ = write_loop(s2).await;
+    });
+
+    println!("Press ctrl-c to stop");
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {
+            println!();
+        }
+    };
+
+    Ok(())
+}
+
+// read_loop shows how to read from the datachannel directly
+#[cfg(feature = "webrtc")]
+async fn read_loop(mut stream: WebRTCStream) -> Result<()> {
+    loop {
+        let Some(res) = stream.next().await else {
+            println!("WebRTC stream closed; Exit the read_loop");
+            return Ok(());
+        };
+        match res {
+            Err(e) => {
+                println!("WebRTC stream read error: {}; Exit the read_loop", e);
+                return Ok(());
+            }
+            Ok(data) => {
+                println!("Message from stream: {}", String::from_utf8(data.to_vec())?);
+            }
+        }
+    }
+}
+
+// write_loop shows how to write to the webrtc stream directly
+#[cfg(feature = "webrtc")]
+async fn write_loop(mut stream: WebRTCStream) -> Result<()> {
+    let mut result = Result::<()>::Ok(());
+    while result.is_ok() {
+        let timeout = tokio::time::sleep(Duration::from_secs(5));
+        tokio::pin!(timeout);
+
+        tokio::select! {
+            _ = timeout.as_mut() =>{
+                let message = webrtc::peer_connection::math_rand_alpha(15);
+                result = stream.send_bytes(Bytes::from(message.clone())).await;
+                println!("Sent '{message}' {}", result.is_ok());
+            }
+        };
+    }
+    println!("WebRTC stream write failed; Exit the write_loop");
+
+    Ok(())
+}

src/config.rs

@@ -2547,6 +2547,7 @@ pub mod keys {
     pub const OPTION_TRACKPAD_SPEED: &str = "trackpad-speed";
     pub const OPTION_REGISTER_DEVICE: &str = "register-device";
     pub const OPTION_RELAY_SERVER: &str = "relay-server";
+    pub const OPTION_ICE_SERVERS: &str = "ice-servers";
     pub const OPTION_DISABLE_UDP: &str = "disable-udp";
     pub const OPTION_ALLOW_INSECURE_TLS_FALLBACK: &str = "allow-insecure-tls-fallback";
     pub const OPTION_SHOW_VIRTUAL_MOUSE: &str = "show-virtual-mouse";
@@ -2743,6 +2744,7 @@ pub mod keys {
         OPTION_ENABLE_ANDROID_SOFTWARE_ENCODING_HALF_SCALE,
         OPTION_ENABLE_TRUSTED_DEVICES,
         OPTION_RELAY_SERVER,
+        OPTION_ICE_SERVERS,
         OPTION_DISABLE_UDP,
         OPTION_ALLOW_INSECURE_TLS_FALLBACK,
     ];

src/lib.rs

@@ -59,6 +59,8 @@ pub mod fingerprint;
 pub use flexi_logger;
 pub mod stream;
 pub mod websocket;
+#[cfg(feature = "webrtc")]
+pub mod webrtc;
 #[cfg(any(target_os = "android", target_os = "ios"))]
 pub use rustls_platform_verifier;
 pub use stream::Stream;

src/socket_client.rs

@@ -1,3 +1,5 @@
+#[cfg(feature = "webrtc")]
+use crate::webrtc::{self, is_webrtc_endpoint};
 use crate::{
     config::{Config, NetworkType},
     tcp::FramedStream,
@@ -129,6 +131,12 @@ pub async fn connect_tcp<
     target: T,
     ms_timeout: u64,
 ) -> ResultType<crate::Stream> {
+    #[cfg(feature = "webrtc")]
+    if is_webrtc_endpoint(&target.to_string()) {
+        return Ok(Stream::WebRTC(
+            webrtc::WebRTCStream::new(&target.to_string(), false, ms_timeout).await?,
+        ));
+    }
     let target_str = check_ws(&target.to_string());
     if is_ws_endpoint(&target_str) {
         return Ok(Stream::WebSocket(

src/stream.rs

@@ -1,10 +1,14 @@
 use crate::{config, tcp, websocket, ResultType};
+#[cfg(feature = "webrtc")]
+use crate::webrtc;
 use sodiumoxide::crypto::secretbox::Key;
 use std::net::SocketAddr;
 use tokio::net::TcpStream;
 
 // support Websocket and tcp.
 pub enum Stream {
+    #[cfg(feature = "webrtc")]
+    WebRTC(webrtc::WebRTCStream),
     WebSocket(websocket::WsFramedStream),
     Tcp(tcp::FramedStream),
 }
@@ -13,6 +17,8 @@ impl Stream {
     #[inline]
     pub fn set_send_timeout(&mut self, ms: u64) {
         match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(s) => s.set_send_timeout(ms),
             Stream::WebSocket(s) => s.set_send_timeout(ms),
             Stream::Tcp(s) => s.set_send_timeout(ms),
         }
@@ -21,6 +27,8 @@ impl Stream {
     #[inline]
     pub fn set_raw(&mut self) {
         match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(s) => s.set_raw(),
             Stream::WebSocket(s) => s.set_raw(),
             Stream::Tcp(s) => s.set_raw(),
         }
@@ -29,6 +37,8 @@ impl Stream {
     #[inline]
     pub async fn send_bytes(&mut self, bytes: bytes::Bytes) -> ResultType<()> {
         match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(s) => s.send_bytes(bytes).await,
             Stream::WebSocket(s) => s.send_bytes(bytes).await,
             Stream::Tcp(s) => s.send_bytes(bytes).await,
         }
@@ -37,6 +47,8 @@ impl Stream {
     #[inline]
     pub async fn send_raw(&mut self, bytes: Vec<u8>) -> ResultType<()> {
         match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(s) => s.send_raw(bytes).await,
             Stream::WebSocket(s) => s.send_raw(bytes).await,
             Stream::Tcp(s) => s.send_raw(bytes).await,
         }
@@ -45,6 +57,8 @@ impl Stream {
     #[inline]
     pub fn set_key(&mut self, key: Key) {
         match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(s) => s.set_key(key),
             Stream::WebSocket(s) => s.set_key(key),
             Stream::Tcp(s) => s.set_key(key),
         }
@@ -53,6 +67,8 @@ impl Stream {
     #[inline]
     pub fn is_secured(&self) -> bool {
         match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(s) => s.is_secured(),
             Stream::WebSocket(s) => s.is_secured(),
             Stream::Tcp(s) => s.is_secured(),
         }
@@ -64,6 +80,8 @@ impl Stream {
         timeout: u64,
     ) -> Option<Result<bytes::BytesMut, std::io::Error>> {
         match self {
+            #[cfg(feature = "webrtc")]
+            Stream::WebRTC(s) => s.next_timeout(timeout).await,
             Stream::WebSocket(s) => s.next_timeout(timeout).await,
             Stream::Tcp(s) => s.next_timeout(timeout).await,
         }
@@ -87,6 +105,8 @@ impl Stream {
     #[inline]
     pub async fn send(&mut self, msg: &impl protobuf::Message) -> ResultType<()> {
         match self {
+            #[cfg(feature = "webrtc")]
+            Self::WebRTC(s) => s.send(msg).await,
             Self::WebSocket(ws) => ws.send(msg).await,
             Self::Tcp(tcp) => tcp.send(msg).await,
         }
@@ -96,6 +116,8 @@ impl Stream {
     #[inline]
     pub async fn next(&mut self) -> Option<Result<bytes::BytesMut, std::io::Error>> {
         match self {
+            #[cfg(feature = "webrtc")]
+            Self::WebRTC(s) => s.next().await,
             Self::WebSocket(ws) => ws.next().await,
             Self::Tcp(tcp) => tcp.next().await,
         }
@@ -104,6 +126,8 @@ impl Stream {
     #[inline]
     pub fn local_addr(&self) -> SocketAddr {
         match self {
+            #[cfg(feature = "webrtc")]
+            Self::WebRTC(s) => s.local_addr(),
             Self::WebSocket(ws) => ws.local_addr(),
             Self::Tcp(tcp) => tcp.local_addr(),
         }
@@ -113,4 +137,13 @@ impl Stream {
     pub fn from(stream: TcpStream, stream_addr: SocketAddr) -> Self {
         Self::Tcp(tcp::FramedStream::from(stream, stream_addr))
     }
+
+    #[inline]
+    #[cfg(feature = "webrtc")]
+    pub fn get_webrtc_stream(&self) -> Option<webrtc::WebRTCStream> {
+        match self {
+            Self::WebRTC(s) => Some(s.clone()),
+            _ => None,
+        }
+    }
 }