Make command line argument program usable as custom node source.

This PR makes it possible to use command line argument as custom node source.
This is very useful when the program is dependant on some environment variable
to be run. Exemples are: ROS/setup.bash and Python/Conda activate.

Custom node source can now be:
- a path: `./target/debut/abcd`
- a program in path: `rosrun`
- a url: `https://github.com/dora-rs/dora/releases/download/v0.0.0-test.4/rust-dataflow-example-node`
- or at the last resort a command line program: `source setup.bash && roslaunch velodyne_pointcloud VLP16_points.launch`

All those sources can have arguments `args` passed through the `args` field of the node. This args field
is recommanded when using a path as it will better manage the different OS.

Cargo.lock

@@ -924,6 +924,7 @@ dependencies = [
  "once_cell",
  "serde",
  "serde_yaml 0.9.11",
+ "which",
  "zenoh-config",
 ]
 

binaries/coordinator/src/run/custom.rs

@@ -1,7 +1,7 @@
 use super::command_init_common_env;
 use dora_core::{
     config::NodeId,
-    descriptor::{self, source_is_url, EnvValue},
+    descriptor::{self, source_is_path, source_is_url, EnvValue},
 };
 use dora_download::download_file;
 use eyre::{eyre, WrapErr};
@@ -15,7 +15,13 @@ pub(super) async fn spawn_custom_node(
     communication: &dora_core::config::CommunicationConfig,
     working_dir: &Path,
 ) -> eyre::Result<tokio::task::JoinHandle<eyre::Result<(), eyre::Error>>> {
-    let path = if source_is_url(&node.source) {
+    let mut is_path = source_is_path(&node.source, working_dir);
+
+    // Use path if node.source correspond to a path
+    let mut command = if let Ok(path) = &is_path {
+        tokio::process::Command::new(path)
+    // Use url if node.source is a url
+    } else if source_is_url(&node.source) {
         // try to download the shared library
         let target_path = Path::new("build")
             .join(node_id.to_string())
@@ -23,25 +29,28 @@ pub(super) async fn spawn_custom_node(
         download_file(&node.source, &target_path)
             .await
             .wrap_err("failed to download custom node")?;
-        target_path
+        is_path = Ok(target_path.clone());
+        tokio::process::Command::new(target_path)
+    // Use node.source as a command if it is not in path and is not a url
+    } else if cfg!(target_os = "windows") {
+        let mut cmd = tokio::process::Command::new("cmd");
+        cmd.args(["/C", &node.source]);
+        cmd
     } else {
-        let raw = Path::new(&node.source);
-        if raw.extension().is_none() {
-            raw.with_extension(EXE_EXTENSION)
-        } else {
-            raw.to_owned()
-        }
-    };
-
-    let cmd = working_dir
-        .join(&path)
-        .canonicalize()
-        .wrap_err_with(|| format!("no node exists at `{}`", path.display()))?;
+        let mut cmd = tokio::process::Command::new("sh");
 
-    let mut command = tokio::process::Command::new(cmd);
+        // Argument are passed in the command string.
+        // Arguments passed after string enclosure will be ignored.
+        cmd.args([
+            "-c",
+            &format!("{} {}", node.source, &node.args.clone().unwrap_or_default()),
+        ]);
+        cmd
+    };
     if let Some(args) = &node.args {
         command.args(args.split_ascii_whitespace());
     }
+
     command_init_common_env(&mut command, &node_id, communication)?;
     command.env(
         "DORA_NODE_RUN_CONFIG",
@@ -59,11 +68,19 @@ pub(super) async fn spawn_custom_node(
     }
 
     let mut child = command.spawn().wrap_err_with(|| {
-        format!(
-            "failed to run executable `{}` with args `{}`",
-            path.display(),
-            node.args.as_deref().unwrap_or_default()
-        )
+        if let Ok(path) = is_path {
+            format!(
+                "failed to run source path: `{}` with args `{}`",
+                path.display(),
+                node.args.as_deref().unwrap_or_default()
+            )
+        } else {
+            format!(
+                "failed to run command: `{}` with args `{}`",
+                node.source,
+                node.args.as_deref().unwrap_or_default()
+            )
+        }
     })?;
     let result = tokio::spawn(async move {
         let status = child.wait().await.context("child process failed")?;

libraries/core/Cargo.toml

@@ -12,3 +12,4 @@ serde = { version = "1.0.136", features = ["derive"] }
 serde_yaml = "0.9.11"
 once_cell = "1.13.0"
 zenoh-config = { git = "https://github.com/eclipse-zenoh/zenoh.git", rev = "79a136e4fd90b11ff5d775ced981af53c4f1071b" }
+which = "4.3.0"

libraries/core/src/descriptor/mod.rs

@@ -1,9 +1,11 @@
 use crate::config::{CommunicationConfig, DataId, InputMapping, NodeId, NodeRunConfig, OperatorId};
+use eyre::{bail, Context, Result};
 use serde::{Deserialize, Serialize};
 use std::{
     collections::{BTreeMap, BTreeSet, HashMap},
+    env::consts::EXE_EXTENSION,
     fmt,
-    path::PathBuf,
+    path::{Path, PathBuf},
 };
 pub use visualize::collect_dora_timers;
 
@@ -176,6 +178,25 @@ pub fn source_is_url(source: &str) -> bool {
     source.contains("://")
 }
 
+pub fn source_is_path(source: &str, working_dir: &Path) -> Result<PathBuf> {
+    let path = Path::new(&source);
+    if path.extension().is_none() {
+        path.with_extension(EXE_EXTENSION)
+    } else {
+        path.to_owned()
+    };
+
+    // Search path within current working directory
+    if let Ok(abs_path) = working_dir.join(&path).canonicalize() {
+        Ok(abs_path)
+    // Search path within $PATH
+    } else if let Ok(abs_path) = which::which(path) {
+        Ok(abs_path)
+    } else {
+        bail!("Could not find source path.")
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct PythonOperatorConfig {
     pub path: PathBuf,