kjuulh/dagger-rs
1
0
Fork 0
mirror of https://github.com/kjuulh/dagger-rs.git synced 2025-07-26 03:19:21 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(core): Fix async panic on blocking #19

Browse Source 
Replaced internal threads with tokio spawn functions
This commit is contained in:
Kasper Juul Hermansen
2023-02-20 10:10:42 +01:00
committed by Kasper Juul Hermansen
parent 45d6462037
commit 75bc17e57d
21 changed files with 66 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
crates/dagger-core/src/cli_session.rs
View File

@@ -3,7 +3,7 @@ use std::{
io::{BufRead, BufReader},
path::PathBuf,
process::{Child, Stdio},
sync::{mpsc::sync_channel, Arc},
sync::Arc,
};
use crate::{config::Config, connect_params::ConnectParams};
@@ -20,12 +20,12 @@ impl CliSession {
}
}
pub fn connect(
pub async fn connect(
&self,
config: &Config,
cli_path: &PathBuf,
) -> eyre::Result<(ConnectParams, Child)> {
self.inner.connect(config, cli_path)
self.inner.connect(config, cli_path).await
}
}
@@ -33,13 +33,13 @@ impl CliSession {
struct InnerCliSession {}
impl InnerCliSession {
pub fn connect(
pub async fn connect(
&self,
config: &Config,
cli_path: &PathBuf,
) -> eyre::Result<(ConnectParams, Child)> {
let proc = self.start(config, cli_path)?;
let params = self.get_conn(proc)?;
let params = self.get_conn(proc).await?;
Ok(params)
}
@@ -70,7 +70,7 @@ impl InnerCliSession {
return Ok(proc);
}
fn get_conn(
async fn get_conn(
&self,
mut proc: std::process::Child,
) -> eyre::Result<(ConnectParams, std::process::Child)> {
@@ -84,14 +84,14 @@ impl InnerCliSession {
.take()
.ok_or(eyre::anyhow!("could not acquire stderr from child process"))?;
let (sender, receiver) = sync_channel(1);
let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
std::thread::spawn(move || {
tokio::spawn(async move {
let stdout_bufr = BufReader::new(stdout);
for line in stdout_bufr.lines() {
let out = line.as_ref().unwrap();
if let Ok(conn) = serde_json::from_str::<ConnectParams>(&out) {
sender.send(conn).unwrap();
sender.send(conn).await.unwrap();
}
if let Ok(line) = line {
println!("dagger: {}", line);
@@ -99,7 +99,7 @@ impl InnerCliSession {
}
});
std::thread::spawn(|| {
tokio::spawn(async move {
let stderr_bufr = BufReader::new(stderr);
for line in stderr_bufr.lines() {
if let Ok(line) = line {
@@ -109,7 +109,7 @@ impl InnerCliSession {
}
});
let conn = receiver.recv()?;
let conn = receiver.recv().await.ok_or(eyre::anyhow!("could not receive ok signal from dagger-engine"))?;
Ok((conn, proc))
}

20
crates/dagger-core/src/downloader.rs
View File

@@ -119,7 +119,7 @@ impl Downloader {
Ok(path)
}
pub fn get_cli(&self) -> eyre::Result<PathBuf> {
pub async fn get_cli(&self) -> eyre::Result<PathBuf> {
let version = &self.version;
let mut cli_bin_path = self.cache_dir()?;
cli_bin_path.push(format!("{CLI_BIN_PREFIX}{version}"));
@@ -129,7 +129,7 @@ impl Downloader {
if !cli_bin_path.exists() {
cli_bin_path = self
.download(cli_bin_path)
.download(cli_bin_path).await
.context("failed to download CLI from archive")?;
}
@@ -145,8 +145,8 @@ impl Downloader {
Ok(cli_bin_path)
}
fn download(&self, path: PathBuf) -> eyre::Result<PathBuf> {
let expected_checksum = self.expected_checksum()?;
async fn download(&self, path: PathBuf) -> eyre::Result<PathBuf> {
let expected_checksum = self.expected_checksum().await?;
let mut bytes = vec![];
let actual_hash = self.extract_cli_archive(&mut bytes)?;
@@ -165,15 +165,15 @@ impl Downloader {
Ok(path)
}
fn expected_checksum(&self) -> eyre::Result<String> {
async fn expected_checksum(&self) -> eyre::Result<String> {
let archive_url = &self.archive_url();
let archive_path = PathBuf::from(&archive_url);
let archive_name = archive_path
.file_name()
.ok_or(eyre::anyhow!("could not get file_name from archive_url"))?;
let resp = reqwest::blocking::get(self.checksum_url())?;
let resp = reqwest::get(self.checksum_url()).await?;
let resp = resp.error_for_status()?;
for line in resp.text()?.lines() {
for line in resp.text().await?.lines() {
let mut content = line.split_whitespace();
let checksum = content
.next()
@@ -240,9 +240,9 @@ impl Downloader {
mod test {
use super::Downloader;
#[test]
fn download() {
let cli_path = Downloader::new("0.3.10".into()).unwrap().get_cli().unwrap();
#[tokio::test]
async fn download() {
let cli_path = Downloader::new("0.3.10".into()).unwrap().get_cli().await.unwrap();
assert_eq!(
Some("dagger-0.3.10"),

16
crates/dagger-core/src/engine.rs
View File

@@ -11,17 +11,17 @@ impl Engine {
Self {}
}
fn from_cli(&self, cfg: &Config) -> eyre::Result<(ConnectParams, Child)> {
let cli = Downloader::new("0.3.12".into())?.get_cli()?;
async fn from_cli(&self, cfg: &Config) -> eyre::Result<(ConnectParams, Child)> {
let cli = Downloader::new("0.3.12".into())?.get_cli().await?;
let cli_session = CliSession::new();
Ok(cli_session.connect(cfg, &cli)?)
Ok(cli_session.connect(cfg, &cli).await?)
}
pub fn start(&self, cfg: &Config) -> eyre::Result<(ConnectParams, Child)> {
pub async fn start(&self, cfg: &Config) -> eyre::Result<(ConnectParams, Child)> {
// TODO: Add from existing session as well
self.from_cli(cfg)
self.from_cli(cfg).await
}
}
@@ -32,10 +32,10 @@ mod tests {
use super::Engine;
// TODO: these tests potentially have a race condition
#[test]
fn engine_can_start() {
#[tokio::test]
async fn engine_can_start() {
let engine = Engine::new();
let params = engine.start(&Config::new(None, None, None, None)).unwrap();
let params = engine.start(&Config::new(None, None, None, None)).await.unwrap();
assert_ne!(
params.0,

12
crates/dagger-core/src/schema.rs
View File

@@ -1,14 +1,14 @@
use crate::introspection::IntrospectionResponse;
use crate::{config::Config, engine::Engine, session::Session};
pub fn get_schema() -> eyre::Result<IntrospectionResponse> {
pub async fn get_schema() -> eyre::Result<IntrospectionResponse> {
let cfg = Config::new(None, None, None, None);
//TODO: Implement context for proc
let (conn, _proc) = Engine::new().start(&cfg)?;
let (conn, _proc) = Engine::new().start(&cfg).await?;
let session = Session::new();
let req_builder = session.start(&cfg, &conn)?;
let schema = session.schema(req_builder)?;
let schema = session.schema(req_builder).await?;
Ok(schema)
}
@@ -17,8 +17,8 @@ pub fn get_schema() -> eyre::Result<IntrospectionResponse> {
mod tests {
use super::get_schema;
#[test]
fn can_get_schema() {
let _ = get_schema().unwrap();
#[tokio::test]
async fn can_get_schema() {
let _ = get_schema().await.unwrap();
}
}

10
crates/dagger-core/src/session.rs
View File

@@ -1,7 +1,7 @@
use graphql_client::GraphQLQuery;
use reqwest::{
blocking::{Client, RequestBuilder},
header::{HeaderMap, HeaderValue, ACCEPT, CONTENT_TYPE},
Client, RequestBuilder,
};
use crate::{config::Config, connect_params::ConnectParams, introspection::IntrospectionResponse};
@@ -37,14 +37,14 @@ impl Session {
Ok(req_builder)
}
pub fn schema(&self, req_builder: RequestBuilder) -> eyre::Result<IntrospectionResponse> {
pub async fn schema(&self, req_builder: RequestBuilder) -> eyre::Result<IntrospectionResponse> {
let request_body: graphql_client::QueryBody<()> = graphql_client::QueryBody {
variables: (),
query: introspection_query::QUERY,
operation_name: introspection_query::OPERATION_NAME,
};
let res = req_builder.json(&request_body).send()?;
let res = req_builder.json(&request_body).send().await?;
if res.status().is_success() {
// do nothing
@@ -52,7 +52,7 @@ impl Session {
return Err(eyre::anyhow!("server error!"));
} else {
let status = res.status();
let error_message = match res.text() {
let error_message = match res.text().await {
Ok(msg) => match serde_json::from_str::<serde_json::Value>(&msg) {
Ok(json) => {
format!("HTTP {}\n{}", status, serde_json::to_string_pretty(&json)?)
@@ -64,7 +64,7 @@ impl Session {
return Err(eyre::anyhow!(error_message));
}
let json: IntrospectionResponse = res.json()?;
let json: IntrospectionResponse = res.json().await?;
Ok(json)
}