WIP: run against dbus-broker bench

Cargo.toml

@@ -23,7 +23,7 @@ path = "src/bin/busd.rs"
 #zbus = { git = "https://github.com/dbus2/zbus/", features = ["tokio"], default-features = false }
 zbus = { version = "3.14.1", features = ["tokio"], default-features = false }
 tokio = { version = "1.19.2", features = ["macros", "rt-multi-thread", "signal", "tracing", "fs" ] }
-clap = { version = "4.0.18", features = ["derive"] }
+clap = { version = "4.3", features = ["derive"] }
 tracing = "0.1.34"
 tracing-subscriber = { version = "0.3.11", features = ["env-filter" , "fmt", "ansi"], default-features = false, optional = true }
 anyhow = "1.0.58"
@@ -36,6 +36,7 @@ console-subscriber = { version = "0.1.8", optional = true }
 hex = "0.4.3"
 xdg-home = "1.0.0"
 rand = "0.8.5"
+quick-xml = { version = "0.29", features = ["serialize", "overlapped-lists"] }
 
 [target.'cfg(unix)'.dependencies]
 nix = "0.26.0"

src/bin/busd.rs

@@ -3,10 +3,10 @@ extern crate busd;
 #[cfg(unix)]
 use std::{fs::File, io::Write, os::fd::FromRawFd};
 
-use busd::bus;
+use busd::{bus, config::BusConfig};
 
 use anyhow::Result;
-use clap::{Parser, ValueEnum};
+use clap::{Args, Parser, ValueEnum};
 #[cfg(unix)]
 use tokio::{select, signal::unix::SignalKind};
 use tracing::error;
@@ -16,7 +16,10 @@ use tracing::{info, warn};
 /// A simple D-Bus broker.
 #[derive(Parser, Debug)]
 #[clap(author, version, about, long_about = None)]
-struct Args {
+struct BusdArgs {
+    #[command(flatten)]
+    config: ConfigArg,
+
     /// The address to listen on.
     #[clap(short = 'a', long, value_parser)]
     address: Option<String>,
@@ -43,6 +46,22 @@ struct Args {
     ready_fd: Option<i32>,
 }
 
+#[derive(Args, Debug)]
+#[group(required = false, multiple = false)]
+struct ConfigArg {
+    /// The configuration file.
+    #[clap(long, value_parser)]
+    config_file: Option<String>,
+
+    /// Use the standard configuration file for the per-login-session message bus.
+    #[clap(long, value_parser)]
+    session: bool,
+
+    /// Use the standard configuration file for the system-wide message bus.
+    #[clap(long, value_parser)]
+    system: bool,
+}
+
 #[derive(Copy, Clone, Debug, ValueEnum)]
 enum AuthMechanism {
     /// This is the recommended authentication mechanism on platforms where credentials can be
@@ -70,10 +89,33 @@ impl From<AuthMechanism> for zbus::AuthMechanism {
 async fn main() -> Result<()> {
     busd::tracing_subscriber::init();
 
-    let args = Args::parse();
+    let mut args = BusdArgs::parse();
+
+    // let's make --session the default
+    if !args.config.session && !args.config.system && args.config.config_file.is_none() {
+        args.config.session = true;
+    }
+    // FIXME: make default config configurable or OS dependant
+    if args.config.session {
+        args.config.config_file = Some("/usr/share/dbus-1/session.conf".into());
+    }
+    if args.config.system {
+        args.config.config_file = Some("/usr/share/dbus-1/system.conf".into());
+    }
+
+    let mut config = BusConfig::default();
+    if let Some(file) = args.config.config_file {
+        config = BusConfig::read(file)?;
+    }
+
+    if let Some(address) = args.address {
+        config.add_listen_address(address);
+    }
 
-    let mut bus =
-        bus::Bus::for_address(args.address.as_deref(), args.auth_mechanism.into()).await?;
+    // FIXME: we don't support multiple <listen> atm
+    let listen_addresses = config.listen_addresses();
+    let address = listen_addresses.last().unwrap();
+    let mut bus = bus::Bus::for_address(address, args.auth_mechanism.into()).await?;
 
     #[cfg(unix)]
     if let Some(fd) = args.ready_fd {

src/bus/mod.rs

@@ -5,12 +5,9 @@ use futures_util::{
     future::{select, Either},
     pin_mut, try_join, TryFutureExt,
 };
-use std::{cell::OnceCell, str::FromStr, sync::Arc};
 #[cfg(unix)]
-use std::{
-    env,
-    path::{Path, PathBuf},
-};
+use std::path::{Path, PathBuf};
+use std::{cell::OnceCell, sync::Arc};
 #[cfg(unix)]
 use tokio::fs::remove_file;
 use tokio::{spawn, task::JoinHandle};
@@ -51,24 +48,53 @@ enum Listener {
 }
 
 impl Bus {
-    pub async fn for_address(address: Option<&str>, auth_mechanism: AuthMechanism) -> Result<Self> {
-        let address = match address {
-            Some(address) => address.to_string(),
-            None => default_address(),
-        };
-        let address = Address::from_str(&address)?;
-        let listener = match &address {
+    pub async fn for_address(address: &str, auth_mechanism: AuthMechanism) -> Result<Self> {
+        let address = address.try_into()?;
+
+        let (listener, address) = match &address {
             #[cfg(unix)]
             Address::Unix(path) => {
                 let path = Path::new(&path);
                 info!("Listening on {}.", path.display());
 
-                Self::unix_stream(path).await
+                Self::unix_stream(path).await.map(|l| (l, address))
+            }
+            #[cfg(unix)]
+            Address::UnixDir(dir) => {
+                let dir = Path::new(&dir);
+                let mut n = 0;
+                loop {
+                    n += 1;
+                    let path = dir.join(format!("dbus-{}", n));
+
+                    let Result::Ok(l) = Self::unix_stream(&path).await else {
+                        continue;
+                    };
+                    info!("Listening on {}.", path.display());
+                    break Ok((l, Address::Unix(path.into())));
+                }
+            }
+            #[cfg(unix)]
+            Address::UnixTmpDir(dir) => {
+                let mut s = std::ffi::OsString::from("\0");
+                s.push(dir);
+                let dir = Path::new(&s);
+                let mut n = 0;
+                loop {
+                    n += 1;
+                    let path = dir.join(format!("dbus-{}", n));
+
+                    let Result::Ok(l) = Self::unix_stream(&path).await else {
+                        continue;
+                    };
+                    info!("Listening on abstract {}.", path.display());
+                    break Ok((l, Address::Unix(path.into())));
+                }
             }
-            Address::Tcp(address) => {
-                info!("Listening on `{}:{}`.", address.host(), address.port());
+            Address::Tcp(tcp) => {
+                info!("Listening on `{}:{}`.", tcp.host(), tcp.port());
 
-                Self::tcp_stream(address).await
+                Self::tcp_stream(tcp).await.map(|l| (l, address))
             }
             Address::NonceTcp { .. } => bail!("`nonce-tcp` transport is not supported (yet)."),
             Address::Autolaunch(_) => bail!("`autolaunch` transport is not supported (yet)."),
@@ -142,6 +168,7 @@ impl Bus {
     #[cfg(unix)]
     async fn unix_stream(socket_path: &Path) -> Result<Listener> {
         let socket_path = socket_path.to_path_buf();
+        let _ = remove_file(&socket_path).await;
         let listener = Listener::Unix {
             listener: tokio::net::UnixListener::bind(&socket_path)?,
             socket_path,
@@ -243,24 +270,3 @@ impl Bus {
         }
     }
 }
-
-#[cfg(unix)]
-fn default_address() -> String {
-    let runtime_dir = env::var("XDG_RUNTIME_DIR")
-        .as_ref()
-        .map(|s| Path::new(s).to_path_buf())
-        .ok()
-        .unwrap_or_else(|| {
-            Path::new("/run")
-                .join("user")
-                .join(format!("{}", nix::unistd::Uid::current()))
-        });
-    let path = runtime_dir.join("busd-session");
-
-    format!("unix:path={}", path.display())
-}
-
-#[cfg(not(unix))]
-fn default_address() -> String {
-    "tcp:host=127.0.0.1,port=4242".to_string()
-}