git.delta.rocks / jrsonnet / refs/commits / 94ece5cae749

1//! This whole library should be replaced with either binding to nix libexpr,2//! or with tvix (once it is able to build NixOS).3//!4//! Current api is awful, little effort was put into this implementation.56use std::{collections::HashMap, path::PathBuf, sync::Arc};78pub use pool::NixSessionPool;9use pool::NixSessionPoolInner;10use r2d2::PooledConnection;11pub use session::{Error, Result};12use tokio::sync::{mpsc, oneshot};13use tracing::instrument;14pub use value::{Index, Value};1516mod pool;17mod session;18mod value;19// Contains macros helpers20#[doc(hidden)]21pub mod macros;22pub mod util;23// #[allow(non_upper_case_globals, non_camel_case_types, non_snake_case)]24// mod nix_raw {25// 	include!(concat!(env!("OUT_DIR"), "/bindings.rs"));26// }2728// fn init() {29// 	nix_raw::libutil_init();30// }3132#[derive(Clone)]33pub struct NixSession(pub(crate) Arc<tokio::sync::Mutex<PooledConnection<NixSessionPoolInner>>>);3435struct NixBuildTask(Value, oneshot::Sender<Result<HashMap<String, PathBuf>>>);3637#[derive(Clone)]38pub struct NixBuildBatch {39	tx: mpsc::UnboundedSender<NixBuildTask>,40}4142#[instrument(skip(session, values))]43async fn build_multiple(name: String, session: NixSession, values: Vec<Value>) -> Result<()> {44	let system = session.0.lock().await.nix_system.clone();45	let builtins = Value::binding(session, "builtins").await?;46	let drv = nix_go!(builtins.derivation(Obj {47		system,48		name,49		builder: "/bin/sh",50		// we want nothing from this derivation, it is only used to perform multiple builds at once.51		args: vec!["-c", "echo > $out"],52		preferLocalBuild: true,53		allowSubstitutes: false,54		buildInputs: values,55	}));56	drv.build().await?;57	Ok(())58}5960impl NixBuildBatch {61	fn new(name: String, session: NixSession) -> Self {62		let (tx, mut rx) = mpsc::unbounded_channel::<NixBuildTask>();6364		tokio::task::spawn(async move {65			let mut deps = vec![];66			let mut build_data = vec![];67			while let Some(task) = rx.recv().await {68				build_data.push(task.0.clone());69				deps.push(task);70			}71			if deps.is_empty() {72				return;73			}74			match build_multiple(name, session, build_data).await {75				Ok(_) => {76					for NixBuildTask(v, o) in deps {77						let _ = o.send(v.build().await);78					}79				}80				Err(e) => {81					for NixBuildTask(v, o) in deps {82						let s = v.to_string_weak().await;83						let s = match s {84							Ok(s) => s,85							Err(e) => {86								let _ = o.send(Err(e));87								continue;88							}89						};90						if PathBuf::from(s).exists() {91							let _ = o.send(v.build().await);92						} else {93							let _ = o.send(Err(e.clone()));94						}95					}96				}97			};98		});99		Self { tx }100	}101	pub async fn submit(self, task: Value) -> Result<HashMap<String, PathBuf>> {102		let Self { tx: task_tx } = self;103		let (tx, rx) = oneshot::channel();104		let _ = task_tx.send(NixBuildTask(task, tx));105		drop(task_tx);106		rx.await.expect("shoudn't be cancelled here")107	}108}109110impl NixSession {111	fn ptr_eq(a: &Self, b: &Self) -> bool {112		Arc::ptr_eq(&a.0, &b.0)113	}114115	pub fn new_build_batch(&self, name: String) -> NixBuildBatch {116		NixBuildBatch::new(name, self.clone())117	}118}119120pub fn init_tokio() {121	let _ = pool::TOKIO_RUNTIME.set(tokio::runtime::Handle::current());122}