fix(all): race condition in process

kjuulh · Feb 20, 2023 · a13a2a9 · a13a2a9
1 parent b86710d
commit a13a2a9
Show file tree

Hide file tree

Showing 4 changed files with 68 additions and 66 deletions.
diff --git a/crates/dagger-codegen/src/rust/templates/object_tmpl.rs b/crates/dagger-codegen/src/rust/templates/object_tmpl.rs
@@ -3,7 +3,7 @@ use genco::prelude::rust;
 use genco::quote;
 use itertools::Itertools;
 
-use crate::functions::{CommonFunctions};
+use crate::functions::CommonFunctions;
 use crate::rust::functions::{
     field_options_struct_name, format_function, format_name, format_optional_args,
     format_struct_comment, format_struct_name,
@@ -12,7 +12,7 @@ use crate::utility::OptionExt;
 
 pub fn render_object(funcs: &CommonFunctions, t: &FullType) -> eyre::Result<rust::Tokens> {
     let selection = rust::import("crate::querybuilder", "Selection");
-    let child = rust::import("std::process", "Child");
+    let child = rust::import("tokio::process", "Child");
     let conn = rust::import("dagger_core::connect_params", "ConnectParams");
     let arc = rust::import("std::sync", "Arc");
 

diff --git a/crates/dagger-core/src/cli_session.rs b/crates/dagger-core/src/cli_session.rs
@@ -6,6 +6,8 @@ use std::{
     sync::Arc,
 };
 
+use tokio::io::AsyncBufReadExt;
+
 use crate::{config::Config, connect_params::ConnectParams};
 
 #[derive(Clone, Debug)]
@@ -24,7 +26,7 @@ impl CliSession {
         &self,
         config: &Config,
         cli_path: &PathBuf,
-    ) -> eyre::Result<(ConnectParams, Child)> {
+    ) -> eyre::Result<(ConnectParams, tokio::process::Child)> {
         self.inner.connect(config, cli_path).await
     }
 }
@@ -37,13 +39,13 @@ impl InnerCliSession {
         &self,
         config: &Config,
         cli_path: &PathBuf,
-    ) -> eyre::Result<(ConnectParams, Child)> {
+    ) -> eyre::Result<(ConnectParams, tokio::process::Child)> {
         let proc = self.start(config, cli_path)?;
         let params = self.get_conn(proc).await?;
         Ok(params)
     }
 
-    fn start(&self, config: &Config, cli_path: &PathBuf) -> eyre::Result<std::process::Child> {
+    fn start(&self, config: &Config, cli_path: &PathBuf) -> eyre::Result<tokio::process::Child> {
         let mut args: Vec<String> = vec!["session".into()];
         if let Some(workspace) = &config.workdir_path {
             let abs_path = canonicalize(workspace)?;
@@ -54,7 +56,7 @@ impl InnerCliSession {
             args.extend(["--project".into(), abs_path.to_string_lossy().to_string()])
         }
 
-        let proc = std::process::Command::new(
+        let proc = tokio::process::Command::new(
             cli_path
                 .to_str()
                 .ok_or(eyre::anyhow!("could not get string from path"))?,
@@ -72,8 +74,8 @@ impl InnerCliSession {
 
     async fn get_conn(
         &self,
-        mut proc: std::process::Child,
-    ) -> eyre::Result<(ConnectParams, std::process::Child)> {
+        mut proc: tokio::process::Child,
+    ) -> eyre::Result<(ConnectParams, tokio::process::Child)> {
         let stdout = proc
             .stdout
             .take()
@@ -87,22 +89,22 @@ impl InnerCliSession {
         let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
 
         tokio::spawn(async move {
-            let stdout_bufr = BufReader::new(stdout);
-            for line in stdout_bufr.lines() {
+            let stdout_bufr = tokio::io::BufReader::new(stdout);
+            for line in stdout_bufr.lines().next_line().await {
                 let out = line.as_ref().unwrap();
                 if let Ok(conn) = serde_json::from_str::<ConnectParams>(&out) {
                     sender.send(conn).await.unwrap();
                 }
-                if let Ok(line) = line {
+                if let Some(line) = line {
                     println!("dagger: {}", line);
                 }
             }
         });
 
         tokio::spawn(async move {
-            let stderr_bufr = BufReader::new(stderr);
-            for line in stderr_bufr.lines() {
-                if let Ok(line) = line {
+            let stdout_bufr = tokio::io::BufReader::new(stderr);
+            for line in stdout_bufr.lines().next_line().await {
+                if let Some(line) = line {
                     println!("dagger: {}", line);
                 }
                 //panic!("could not start dagger session: {}", out)

diff --git a/crates/dagger-core/src/engine.rs b/crates/dagger-core/src/engine.rs
@@ -11,15 +11,15 @@ impl Engine {
         Self {}
     }
 
-    async fn from_cli(&self, cfg: &Config) -> eyre::Result<(ConnectParams, Child)> {
+    async fn from_cli(&self, cfg: &Config) -> eyre::Result<(ConnectParams, tokio::process::Child)> {
         let cli = Downloader::new("0.3.12".into())?.get_cli().await?;
 
         let cli_session = CliSession::new();
 
         Ok(cli_session.connect(cfg, &cli).await?)
     }
 
-    pub async fn start(&self, cfg: &Config) -> eyre::Result<(ConnectParams, Child)> {
+    pub async fn start(&self, cfg: &Config) -> eyre::Result<(ConnectParams, tokio::process::Child)> {
         // TODO: Add from existing session as well
         self.from_cli(cfg).await
     }