replace tide with axum in rest plugin and remove async_std dependencies

Cargo.toml

@@ -88,6 +88,7 @@ async-global-executor = "3.1.0"
 async-io = "2.6.0"
 async-std = { version = "1.13.2", features = ["tokio1"] }
 async-trait = "0.1.89"
+axum = "0.8.4"
 base64 = "0.22.1"
 bincode = "1.3.3"
 buddy_system_allocator = "0.10.0"
@@ -113,7 +114,6 @@ hashbrown = "0.16.0"
 hex = { version = "0.4.3", default-features = false } # Default features are disabled due to usage in no_std crates
 hmac = { version = "0.12.1", features = ["std"] }
 home = "0.5.9"
-http-types = "2.12.0"
 humantime = "2.3.0"
 itertools = "0.14.0"
 json5 = "0.4.1"
@@ -123,7 +123,8 @@ lazy_static = "1.5.0"
 leb128 = "0.2"
 libc = "0.2.175"
 libloading = "0.8.9"
-lz4_flex = "0.10.0"
+lz4_flex = "0.11.3"
+mime = "0.3.17"
 nix = { version = "0.29.0", features = ["fs"] }
 nonempty-collections = { version = "0.3.1", features = ["serde"] }
 num-traits = { version = "0.2.19", default-features = false }
@@ -176,14 +177,15 @@ stop-token = "0.7.0"
 syn = "2.0.110"
 talc = { version = "4.4.3", default-features = false }
 test-case = "3.3.1"
-tide = "0.16.0"
 time = "0.3.41"
 token-cell = { version = "2.0.0", default-features = false }
 tokio = { version = "1.47.1", default-features = false } # Default features are disabled due to some crates' requirements
 tokio-rustls = { version = "0.26.2", default-features = false }
 tokio-tungstenite = "0.24.0"
 tokio-util = "0.7.16"
 toml = "0.9.6"
+tower = "0.5.2"
+tower-http = "0.6.6"
 tracing = "0.1.41"
 tracing-capture = "0.1.0"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }

commons/zenoh-core/src/macros.rs

@@ -156,27 +156,6 @@ macro_rules! to_u64 {
     };
 }
 
-// This macro allows to spawn the right amount of threads in the
-// async_std executor
-#[macro_export]
-macro_rules! zasync_executor_init {
-    () => {
-        use async_global_executor;
-
-        // Zenoh requires at least 4 threads to run
-        const ASYNC_STD_THREAD_COUNT_MIN: usize = 4;
-
-        let count = async_global_executor::spawn_more_threads(ASYNC_STD_THREAD_COUNT_MIN)
-            .await
-            .unwrap();
-
-        tracing::trace!(
-            "Spawned {} additional threads in the async global executor",
-            count
-        );
-    };
-}
-
 // This macro allows to parse a string to the target type
 #[macro_export]
 macro_rules! zparse {

commons/zenoh-pinned-deps-1-75/Cargo.toml

@@ -27,7 +27,8 @@ version = { workspace = true }
 maintenance = { status = "actively-developed" }
 
 [dependencies]
-async-lock = "=3.4.1"
+axum = "=0.8.4"
+axum-core = "=0.5.2"
 base64ct = "=1.6.0"
 clap_builder = "=4.5.60"
 clap_lex = "=1.0.0"
@@ -72,7 +73,8 @@ security-framework = "=3.6.0"
 
 [package.metadata.cargo-machete]
 ignored = [
-  "async-lock",
+  "axum",
+  "axum-core",
   "base64ct",
   "clap_builder",
   "clap_lex",

plugins/zenoh-plugin-rest/Cargo.toml

@@ -26,26 +26,26 @@ version = { workspace = true }
 [features]
 default = ["dynamic_plugin", "zenoh/default"]
 dynamic_plugin = []
-static_plugin = ['async-std']
+static_plugin = []
 
 [lib]
 crate-type = ["cdylib", "rlib"]
 name = "zenoh_plugin_rest"
 
 [dependencies]
-anyhow = { workspace = true, features = ["default"] }
-async-std = { workspace = true, features = ["tokio1"], optional = true }
+axum = { workspace = true, features = ["tokio"] }
 base64 = { workspace = true }
 flume = { workspace = true }
 futures = { workspace = true }
 git-version = { workspace = true }
-http-types = { workspace = true }
 lazy_static = { workspace = true }
+mime = { workspace = true }
 schemars = { workspace = true }
 serde = { workspace = true, features = ["default"] }
 serde_json = { workspace = true }
-tide = { workspace = true }
 tokio = { workspace = true }
+tower = { workspace = true }
+tower-http = { workspace = true, features = ["cors", "set-header", "trace"] }
 tracing = { workspace = true }
 zenoh = { workspace = true, default-features = false, features = [
   "internal",
@@ -75,6 +75,3 @@ license-file = ["../../LICENSE", "0"]
 maintainer = "zenoh-dev@eclipse.org"
 name = "zenoh-plugin-rest"
 section = "net"
-
-[package.metadata.cargo-machete]
-ignored = ["async-std"]

plugins/zenoh-plugin-rest/examples/z_serve_sse.rs

@@ -22,6 +22,7 @@ const HTML: &str = r#"
 if(typeof(EventSource) !== "undefined") {
   var source = new EventSource("/demo/sse/event");
   source.addEventListener("PUT", function(e) {
+    console.log(e);
     document.getElementById("result").innerHTML += e.data + "<br>";
   }, false);
 } else {
@@ -89,26 +90,26 @@ fn parse_args() -> Config {
         )
         .get_matches();
 
-    let mut config = if let Some(conf_file) = args.get_one::<&String>("config") {
+    let mut config = if let Some(conf_file) = args.get_one::<String>("config") {
         Config::from_file(conf_file).unwrap()
     } else {
         Config::default()
     };
-    match args.get_one::<&String>("mode").map(|m| m.parse()) {
+    match args.get_one::<String>("mode").map(|m| m.parse()) {
         Some(Ok(mode)) => {
             config.set_mode(Some(mode)).unwrap();
         }
         Some(Err(e)) => panic!("Invalid mode: {e}"),
         None => {}
     };
-    if let Some(values) = args.get_many::<&String>("connect") {
+    if let Some(values) = args.get_many::<String>("connect") {
         config
             .connect
             .endpoints
             .set(values.into_iter().map(|v| v.parse().unwrap()).collect())
             .unwrap();
     }
-    if let Some(values) = args.get_many::<&String>("listen") {
+    if let Some(values) = args.get_many::<String>("listen") {
         config
             .listen
             .endpoints

plugins/zenoh-plugin-rest/src/config.rs

@@ -11,24 +11,22 @@
 // Contributors:
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
-use std::fmt;
+use std::{
+    fmt,
+    net::{Ipv6Addr, SocketAddr},
+};
 
 use schemars::JsonSchema;
-use serde::{
-    de,
-    de::{Unexpected, Visitor},
-    Deserialize, Deserializer,
-};
+use serde::{de, de::Visitor, Deserialize, Deserializer};
 
-const DEFAULT_HTTP_INTERFACE: &str = "[::]";
 pub const DEFAULT_WORK_THREAD_NUM: usize = 2;
 pub const DEFAULT_MAX_BLOCK_THREAD_NUM: usize = 50;
 
 #[derive(JsonSchema, Deserialize, serde::Serialize, Clone, Debug)]
 #[serde(deny_unknown_fields)]
 pub struct Config {
-    #[serde(deserialize_with = "deserialize_http_port")]
-    pub http_port: String,
+    #[serde(rename = "http_port", deserialize_with = "deserialize_http_port")]
+    pub addr: SocketAddr,
     #[serde(default = "default_work_thread_num")]
     pub work_thread_num: usize,
     #[serde(default = "default_max_block_thread_num")]
@@ -46,7 +44,7 @@ impl From<&Config> for serde_json::Value {
     }
 }
 
-fn deserialize_http_port<'de, D>(deserializer: D) -> Result<String, D::Error>
+fn deserialize_http_port<'de, D>(deserializer: D) -> Result<SocketAddr, D::Error>
 where
     D: Deserializer<'de>,
 {
@@ -64,7 +62,7 @@ fn default_max_block_thread_num() -> usize {
 struct HttpPortVisitor;
 
 impl Visitor<'_> for HttpPortVisitor {
-    type Value = String;
+    type Value = SocketAddr;
 
     fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
         formatter.write_str(r#"either a port number as an integer or a string, either a string with format "<local_ip>:<port_number>""#)
@@ -74,26 +72,30 @@ impl Visitor<'_> for HttpPortVisitor {
     where
         E: de::Error,
     {
-        Ok(format!("{DEFAULT_HTTP_INTERFACE}:{value}"))
+        Ok(SocketAddr::new(
+            Ipv6Addr::UNSPECIFIED.into(),
+            value
+                .try_into()
+                .map_err(|_| de::Error::custom("invalid port"))?,
+        ))
     }
 
     fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
     where
         E: de::Error,
     {
-        let parts: Vec<&str> = value.split(':').collect();
-        if parts.len() > 2 {
-            return Err(E::invalid_value(Unexpected::Str(value), &self));
-        }
-        let (interface, port) = if parts.len() == 1 {
-            (DEFAULT_HTTP_INTERFACE, parts[0])
-        } else {
-            (parts[0], parts[1])
+        let (ip, port) = match value.split_once(':') {
+            Some((ip, port)) => (
+                ip.parse()
+                    .map_err(|_| de::Error::custom("invalid ip address"))?,
+                port,
+            ),
+            None => (Ipv6Addr::UNSPECIFIED.into(), value),
         };
-        if port.parse::<u32>().is_err() {
-            return Err(E::invalid_value(Unexpected::Str(port), &self));
-        }
-        Ok(format!("{interface}:{port}"))
+        let port = port
+            .parse()
+            .map_err(|_| de::Error::custom("invalid port"))?;
+        Ok(SocketAddr::new(ip, port))
     }
 }
 
@@ -159,7 +161,7 @@ impl<'de> serde::de::Visitor<'de> for PathVisitor {
 
 #[cfg(test)]
 mod tests {
-    use super::{Config, DEFAULT_HTTP_INTERFACE};
+    use super::Config;
 
     #[test]
     fn test_path_field() {
@@ -169,13 +171,13 @@ mod tests {
 
         assert!(config.is_ok());
         let Config {
-            http_port,
+            addr: http_port,
             __required__,
             __path__,
             ..
         } = config.unwrap();
 
-        assert_eq!(http_port, format!("{DEFAULT_HTTP_INTERFACE}:8080"));
+        assert_eq!(http_port.to_string(), "[::]:8080");
         assert_eq!(__path__, Some(vec![String::from("/example/path")]));
         assert_eq!(__required__, None);
     }
@@ -186,13 +188,13 @@ mod tests {
         let config = serde_json::from_str::<Config>(r#"{"__required__": true, "http_port": 8080}"#);
         assert!(config.is_ok());
         let Config {
-            http_port,
+            addr: http_port,
             __required__,
             __path__,
             ..
         } = config.unwrap();
 
-        assert_eq!(http_port, format!("{DEFAULT_HTTP_INTERFACE}:8080"));
+        assert_eq!(http_port.to_string(), "[::]:8080");
         assert_eq!(__path__, None);
         assert_eq!(__required__, Some(true));
     }
@@ -206,13 +208,13 @@ mod tests {
 
         assert!(config.is_ok());
         let Config {
-            http_port,
+            addr: http_port,
             __required__,
             __path__,
             ..
         } = config.unwrap();
 
-        assert_eq!(http_port, format!("{DEFAULT_HTTP_INTERFACE}:8080"));
+        assert_eq!(http_port.to_string(), "[::]:8080");
         assert_eq!(__path__, Some(vec![String::from("/example/path")]));
         assert_eq!(__required__, Some(true));
     }
@@ -224,13 +226,13 @@ mod tests {
 
         assert!(config.is_ok());
         let Config {
-            http_port,
+            addr: http_port,
             __required__,
             __path__,
             ..
         } = config.unwrap();
 
-        assert_eq!(http_port, format!("{DEFAULT_HTTP_INTERFACE}:8080"));
+        assert_eq!(http_port.to_string(), "[::]:8080");
         assert_eq!(__path__, None);
         assert_eq!(__required__, None);
     }