worldline · reneca · Jan 2, 2025 · Jan 9, 2025
diff --git a/Cargo.toml b/Cargo.toml
@@ -24,11 +24,14 @@ include = [
 [workspace.dependencies]
 prosa-utils = { version = "0.1.2", path = "prosa_utils" }
 prosa-macros = { version = "0.1.2", path = "prosa_macros" }
-thiserror = "1"
-aquamarine = "0.5"
+thiserror = "2"
+aquamarine = "0.6"
 bytes = "1"
+chrono = "0.4"
 url = { version = "2", features = ["serde"] }
-tokio = "1"
+tokio = { version = "1", features = ["fs", "macros", "net", "parking_lot", "rt", "rt-multi-thread", "signal", "sync", "time"] }
+config = { version = "0.15", default-features = false, features = ["toml", "json", "yaml", "json5", "convert-case", "async"] }
+toml = "0.8"
 
 # Config Observability
 log = "0.4"

diff --git a/cargo-prosa/Cargo.toml b/cargo-prosa/Cargo.toml
@@ -14,20 +14,18 @@ name = "cargo-prosa"
 path = "src/main.rs"
 
 [dependencies]
-bytes.workspace = true
 thiserror.workspace = true
 prosa-utils = { workspace = true, features = ["msg", "config", "config-observability"] }
 aquamarine.workspace = true
-clap = { version = "4", features = ["derive"] }
+clap = "4"
 clap_complete = "4"
 serde = "1"
-toml = "0.8"
+toml.workspace = true
 toml_edit = { version = "0.22", features = ["serde"] }
 serde_json = "1"
 serde_yaml = "0.9"
-config = "0.13"
+config.workspace = true
 tera = "1"
-daemonize = "0.5"
 
 [dev-dependencies]
 assert_cmd = "2"

diff --git a/cargo-prosa/assets/build.rs.j2 b/cargo-prosa/assets/build.rs.j2
@@ -11,7 +11,7 @@ use cargo_prosa::CONFIGURATION_FILENAME;
 use cargo_prosa::package::deb::DebPkg;
 {% endif %}
 
-fn write_settings_rs(out_dir: &OsString, desc: &Desc, metadata: &HashMap<String, Metadata>) -> io::Result<()> {{ '{' }}
+fn write_settings_rs(out_dir: &OsString, desc: &Desc, metadata: &HashMap<&str, Metadata>) -> io::Result<()> {{ '{' }}
     let mut f = fs::File::create(Path::new(&out_dir).join("settings.rs"))?;
     writeln!(f, "use prosa::core::settings::settings;")?;
 
@@ -21,7 +21,7 @@ fn write_settings_rs(out_dir: &OsString, desc: &Desc, metadata: &HashMap<String,
         writeln!(f, "#[derive(Default, Debug, Deserialize, Serialize)]")?;
         writeln!(f, "pub struct RunSettings {{ '{{' }}")?;
         for processor in processors {{ '{' }}
-            let proc_metadata = metadata.get(&processor.proc_name).unwrap_or_else(|| panic!("Can't get the processor {{ '{}' }} metadata ({{ '{:?}' }})", processor.proc, processor.name));
+            let proc_metadata = metadata.get(processor.proc_name.as_str()).unwrap_or_else(|| panic!("Can't get the processor {{ '{}' }} metadata ({{ '{:?}' }})", processor.proc, processor.name));
             if let Some(settings) = &proc_metadata.settings {{ '{' }}
                 if let Some(description) = &proc_metadata.description {{ '{' }}
                     writeln!(f, "    /// {{ '{}' }}", description)?;
@@ -100,7 +100,7 @@ fn write_config_rs(out_dir: &OsString, desc: &Desc, cargo_metadata: &CargoMetada
     writeln!(f, "{{ '}}' }}\n")
 {{ '}' }}
 
-fn write_run_rs(out_dir: &OsString, desc: &Desc, metadata: &HashMap<String, Metadata>) -> io::Result<()> {{ '{' }}
+fn write_run_rs(out_dir: &OsString, desc: &Desc, metadata: &HashMap<&str, Metadata>) -> io::Result<()> {{ '{' }}
     let mut f = fs::File::create(Path::new(&out_dir).join("run.rs"))?;
 
     writeln!(f, "fn new_main(settings: &RunSettings) -> (prosa::core::main::Main<{{ '{}' }}>, {{ '{}' }}<{{ '{}' }}>) {{ '{{' }}", desc.prosa.tvf, desc.prosa.main, desc.prosa.tvf)?;
@@ -115,7 +115,7 @@ fn write_run_rs(out_dir: &OsString, desc: &Desc, metadata: &HashMap<String, Meta
         for processor in processors {{ '{' }}
             proc_id += 1;
             writeln!(f, "debug!(\"Start processor {{ '{}' }}\");", processor.get_name())?;
-            let proc_metadata = metadata.get(&processor.proc_name).unwrap_or_else(|| panic!("Can't get the processor {{ '{}' }} metadata ({{ '{:?}' }})", processor.proc, processor.name));
+            let proc_metadata = metadata.get(processor.proc_name.as_str()).unwrap_or_else(|| panic!("Can't get the processor {{ '{}' }} metadata ({{ '{:?}' }})", processor.proc, processor.name));
             if proc_metadata.settings.is_some() {{ '{' }}
                 writeln!(f, "let proc = {{ '{}' }}::<{{ '{}' }}>::create({{ '{}' }}, bus.clone(), settings.{{ '{}' }}.clone());", processor.proc, desc.prosa.tvf, proc_id, processor.get_name().replace('-', "_"))?;
             {{ '}' }} else {{ '{' }}
@@ -186,8 +186,8 @@ fn write_target_config(out_dir: &OsString, target_dir: &Path) -> io::Result<()>
 
     // Correct relative path in the Cargo.toml by the `out_dir` one
     let cargo_content = fs::read_to_string("Cargo.toml")?.replace("\"../", format!("\"{}/../", env::current_dir()?.display()).as_str());
-    let mut cargo_dst = fs::File::create(&prosa_config_path.join("Cargo.toml"))?;
-    cargo_dst.write(cargo_content.as_bytes())?;
+    let mut cargo_dst = fs::File::create(prosa_config_path.join("Cargo.toml"))?;
+    cargo_dst.write_all(cargo_content.as_bytes())?;
 
     let mut f = fs::File::create(prosa_config_path.join("src").join("main.rs"))?;
     writeln!(f, "use prosa::core::settings::Settings;\n")?;

diff --git a/cargo-prosa/src/builder.rs b/cargo-prosa/src/builder.rs
@@ -194,13 +194,16 @@ impl Desc {
     where
         P: AsRef<Path>,
     {
-        let mut toml_file = fs::File::create(&path)?;
-        writeln!(toml_file, "# ProSA definition")?;
-        writeln!(
-            toml_file,
-            "{}",
-            toml::to_string(&self).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?
-        )
+        fn inner(desc: &Desc, mut file: fs::File) -> Result<(), io::Error> {
+            writeln!(file, "# ProSA definition")?;
+            writeln!(
+                file,
+                "{}",
+                toml::to_string(&desc)
+                    .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?
+            )
+        }
+        inner(self, fs::File::create(&path)?)
     }
 
     /// Method to read a ProSA toml description file

diff --git a/cargo-prosa/src/cargo.rs b/cargo-prosa/src/cargo.rs
@@ -39,9 +39,9 @@ pub struct ComponentVersion<'a> {
     /// Name of the component
     pub name: String,
     /// Name of the component's crate
-    pub crate_name: &'a String,
+    pub crate_name: &'a str,
     /// Version of the component
-    pub version: &'a String,
+    pub version: &'a str,
 }
 
 impl fmt::Display for ComponentVersion<'_> {
@@ -61,7 +61,7 @@ impl fmt::Display for ComponentVersion<'_> {
 }
 
 /// Metadata of a ProSA processor
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Clone, Deserialize)]
 pub struct Metadata {
     /// Name of the ProSA package (not included in the metadata)
     #[serde(skip_deserializing)]
@@ -249,7 +249,7 @@ impl PackageMetadata {
     }
 
     /// Getter of the ProSA Processor or Adaptor if present
-    pub fn get_prosa_proc_metadata(&self) -> Option<HashMap<String, Metadata>> {
+    pub fn get_prosa_proc_metadata(&self) -> Option<HashMap<&str, Metadata>> {
         if let Some(metadata) = self
             .metadata
             .as_ref()
@@ -259,7 +259,7 @@ impl PackageMetadata {
             for (name, data) in metadata {
                 if name != "main" && name != "tvf" {
                     if let Ok(prosa_metadata) = serde_json::from_value::<Metadata>(data.clone()) {
-                        proc_metadata.insert(name.clone(), prosa_metadata);
+                        proc_metadata.insert(name.as_str(), prosa_metadata);
                     }
                 }
             }
@@ -489,13 +489,13 @@ impl CargoMetadata {
     }
 
     /// Method to get all merged ProSA proc metadata
-    pub fn prosa_proc_metadata(&self) -> HashMap<String, Metadata> {
-        let mut prosa_list: HashMap<String, Metadata> = HashMap::with_capacity(self.packages.len());
+    pub fn prosa_proc_metadata(&self) -> HashMap<&str, Metadata> {
+        let mut prosa_list: HashMap<&str, Metadata> = HashMap::with_capacity(self.packages.len());
         for package in &self.packages {
             if let Some(prosa_metadata) = package.get_prosa_proc_metadata() {
                 for (prosa_proc_name, mut prosa_proc_metadata) in prosa_metadata {
                     prosa_proc_metadata.specify(&package.name, package.description.clone());
-                    if let Some(prosa_existing_metadata) = prosa_list.get_mut(&prosa_proc_name) {
+                    if let Some(prosa_existing_metadata) = prosa_list.get_mut(prosa_proc_name) {
                         prosa_existing_metadata.merge(prosa_proc_metadata);
                     } else {
                         prosa_list.insert(prosa_proc_name, prosa_proc_metadata);

diff --git a/cargo-prosa/src/main.rs b/cargo-prosa/src/main.rs
@@ -62,31 +62,35 @@ fn render_build_rs<P>(path: P, ctx: &tera::Context) -> Result<(), tera::Error>
 where
     P: AsRef<Path>,
 {
-    const RENDER_FILENAME: &str = "build.rs";
-    let mut tera_build = Tera::default();
-    tera_build.add_raw_template(
-        RENDER_FILENAME,
-        include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/assets/build.rs.j2")),
-    )?;
-
-    let build_file = fs::File::create(&path).map_err(tera::Error::io_error)?;
-    tera_build.render_to(RENDER_FILENAME, ctx, build_file)
+    fn inner(file: fs::File, ctx: &tera::Context) -> Result<(), tera::Error> {
+        const RENDER_FILENAME: &str = "build.rs";
+        let mut tera_build = Tera::default();
+        tera_build.add_raw_template(
+            RENDER_FILENAME,
+            include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/assets/build.rs.j2")),
+        )?;
+
+        tera_build.render_to(RENDER_FILENAME, ctx, file)
+    }
+    inner(fs::File::create(&path).map_err(tera::Error::io_error)?, ctx)
 }
 
 /// Function to render jinja build.rs file into prosa project
 fn render_main_rs<P>(path: P, ctx: &tera::Context) -> Result<(), tera::Error>
 where
     P: AsRef<Path>,
 {
-    const RENDER_FILENAME: &str = "main.rs";
-    let mut tera_build = Tera::default();
-    tera_build.add_raw_template(
-        RENDER_FILENAME,
-        include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/assets/main.rs.j2")),
-    )?;
-
-    let main_file = fs::File::create(&path).map_err(tera::Error::io_error)?;
-    tera_build.render_to(RENDER_FILENAME, ctx, main_file)
+    fn inner(file: fs::File, ctx: &tera::Context) -> Result<(), tera::Error> {
+        const RENDER_FILENAME: &str = "main.rs";
+        let mut tera_build = Tera::default();
+        tera_build.add_raw_template(
+            RENDER_FILENAME,
+            include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/assets/main.rs.j2")),
+        )?;
+
+        tera_build.render_to(RENDER_FILENAME, ctx, file)
+    }
+    inner(fs::File::create(&path).map_err(tera::Error::io_error)?, ctx)
 }
 
 /// Function to initiate ProSA project file (or update them if existing)
@@ -98,9 +102,21 @@ fn init_prosa(path: &str, context: &tera::Context) -> io::Result<()> {
     let cargo_add_prosa_utils = cargo!("add", Some(path), "prosa-utils");
     let cargo_add_clap = cargo!("add", Some(path), "clap");
     let cargo_add_daemonize = cargo!("add", Some(path), "daemonize");
-    let cargo_add_tokio = cargo!("add", Some(path), "tokio");
+    let cargo_add_tokio = cargo!(
+        "add",
+        Some(path),
+        "tokio",
+        "-F",
+        "macros,rt,rt-multi-thread"
+    );
     let cargo_add_serde = cargo!("add", Some(path), "serde");
-    let cargo_add_config = cargo!("add", Some(path), "config");
+    let cargo_add_config = cargo!(
+        "add",
+        Some(path),
+        "config",
+        "-F",
+        "toml,json,yaml,json5,convert-case,async"
+    );
     let cargo_add_tracing = cargo!("add", Some(path), "tracing");
 
     // Add build dependencies
@@ -345,7 +361,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
                 if let Some(processor) = matches.get_one::<String>("PROCESSOR") {
                     if let Some(proc_metadata) = CargoMetadata::load_metadata()?
                         .prosa_proc_metadata()
-                        .get(processor)
+                        .get(processor.as_str())
                     {
                         let mut proc_desc = proc_metadata.get_proc_desc(
                             matches.get_one::<String>("adaptor").map(|x| x.as_str()),

diff --git a/prosa/Cargo.toml b/prosa/Cargo.toml
@@ -28,9 +28,8 @@ adaptor = ["stub::adaptor::StubParotAdaptor"]
 
 [dependencies]
 prosa-utils = { workspace = true, features = ["msg", "config", "config-observability"] }
-prosa-macros = { workspace = true }
-bytes = {workspace = true}
-chrono= "0.4"
+prosa-macros.workspace = true
+bytes.workspace = true
 tracing = "0.1"
 tracing-subscriber = {version = "0.3", features = ["std", "env-filter"]}
 thiserror.workspace = true
@@ -40,14 +39,14 @@ rlimit = "0.10"
 aquamarine.workspace = true
 
 openssl = { version = "0.10" }
-tokio = { workspace = true, features = ["full"] }
+tokio.workspace = true
 tokio-openssl = "0.6"
 async-http-proxy = { version = "1", features = ["runtime-tokio","basic-auth"] }
 
 serde = { version = "1", features = ["derive"] }
-config = "0.13"
+config.workspace = true
 glob = { version = "0.3" }
-toml = "0.8"
+toml.workspace = true
 serde_yaml = "0.9"
 
 log.workspace = true

diff --git a/prosa/src/core/main.rs b/prosa/src/core/main.rs
@@ -583,20 +583,35 @@ where
     M: Sized + Clone + Debug + Tvf + Default + 'static + std::marker::Send + std::marker::Sync,
 {
     fn create<S: Settings>(settings: &S) -> (Main<M>, MainProc<M>) {
+        fn inner<M>(
+            main: Main<M>,
+            internal_rx_queue: mpsc::Receiver<InternalMainMsg<M>>,
+        ) -> (Main<M>, MainProc<M>)
+        where
+            M: Sized
+                + Clone
+                + Debug
+                + Tvf
+                + Default
+                + 'static
+                + std::marker::Send
+                + std::marker::Sync,
+        {
+            let name = main.name().clone();
+            let meter = main.meter("prosa_main_task_meter");
+            (
+                main,
+                MainProc {
+                    name,
+                    processors: Default::default(),
+                    services: Arc::new(ServiceTable::default()),
+                    internal_rx_queue,
+                    meter,
+                },
+            )
+        }
         let (internal_tx_queue, internal_rx_queue) = mpsc::channel(2048);
-        let main = Main::new(internal_tx_queue, settings);
-        let name = main.name().clone();
-        let meter = main.meter("prosa_main_task_meter");
-        (
-            main,
-            MainProc {
-                name,
-                processors: Default::default(),
-                services: Arc::new(ServiceTable::default()),
-                internal_rx_queue,
-                meter,
-            },
-        )
+        inner(Main::new(internal_tx_queue, settings), internal_rx_queue)
     }
 
     fn run(mut self) -> std::thread::JoinHandle<()> {

diff --git a/prosa_macros/Cargo.toml b/prosa_macros/Cargo.toml
@@ -13,11 +13,11 @@ include.workspace = true
 proc-macro = true
 
 [dependencies]
-syn = { version = "2", features = ["full", "extra-traits"] }
+syn = "2"
 quote = "1"
 proc-macro2 = "1"
-chrono = "0.4"
+chrono.workspace = true
 
 [dev-dependencies]
-bytes = { workspace = true }
-prosa-utils = { workspace = true }
+bytes.workspace = true
+prosa-utils.workspace = true
diff --git a/prosa_utils/Cargo.toml b/prosa_utils/Cargo.toml
@@ -12,7 +12,7 @@ include.workspace = true
 [features]
 default = ["full"]
 msg = []
-config = ["dep:glob","dep:serde","dep:toml","dep:serde_yaml"]
+config = ["dep:glob","dep:serde","dep:serde_yaml"]
 config-openssl = ["config", "dep:openssl"]
 config-observability = ["dep:log", "dep:tracing-core", "dep:tracing-subscriber", "dep:tracing-opentelemetry", "dep:opentelemetry", "dep:opentelemetry_sdk", "dep:opentelemetry-stdout", "dep:opentelemetry-otlp"]
 config-observability-prometheus = ["config-observability", "dep:prometheus", "dep:prometheus_exporter", "dep:opentelemetry-prometheus"]
@@ -25,13 +25,12 @@ tvf = ["msg::simple_string_tvf::SimpleStringTvf"]
 thiserror.workspace = true
 bytes.workspace = true
 url.workspace = true
-chrono = "0.4"
+chrono.workspace = true
 hex = "0.4"
 
 # Config
 glob = { version = "0.3", optional = true }
 serde = { version = "1", optional = true, features = ["derive"] }
-toml = { version = "0.8", optional = true }
 serde_yaml = { version = "0.9", optional = true }
 
 # Config OpenSSL
@@ -51,5 +50,5 @@ prometheus_exporter = { workspace = true, optional = true }
 opentelemetry-prometheus = { workspace = true, optional = true }
 
 [dev-dependencies]
-tokio = { workspace = true, features = ["full"] }
+tokio.workspace = true
 tokio-openssl = "0.6"