fix: made functions sendable between threads

Cargo.lock

@@ -3388,6 +3388,7 @@ version = "0.2.0-alpha.2"
 dependencies = [
  "anyhow",
  "async-compression",
+ "async-stream",
  "async-trait",
  "async_zip",
  "base64 0.22.1",

Cargo.toml

@@ -160,6 +160,7 @@ sea-orm-migration = { version = "1.1", features = [
 # utils
 anyhow = { version = "1" }
 async-trait = { version = "=0.1.88" }
+async-stream = { version = "=0.3.6" }
 cc = { version = "1" }
 discord-rich-presence = { version = "=0.2.5" }
 futures = { version = "0.3" }

apps/oneclient/frontend/src/bindings.gen.ts

@@ -42,8 +42,6 @@ export type JavaError = { ParseVersion: [string] } | 'Execute' | 'MissingJava';
 
 export type LauncherError = { type: 'DirError'; data: DirectoryError } | { type: 'IOError'; data: IOError } | { type: 'IngressError'; data: IngressError } | { type: 'JavaError'; data: JavaError } | { type: 'CryptoError'; data: CryptoError } | { type: 'DiscordError'; data: DiscordError } | { type: 'MetadataError'; data: MetadataError } | { type: 'ClusterError'; data: ClusterError } | { type: 'MinecraftAuthError'; data: MinecraftAuthError } | { type: 'ProcessError'; data: ProcessError } | { type: 'PackageError'; data: PackageError } | { type: 'DaoError'; data: DaoError } | { type: 'SerdeError' } | { type: 'AnyhowError' } | { type: 'DbError' } | { type: 'ReqwestError' } | { type: 'InterpulseError' } | { type: 'RegexError' } | { type: 'SemaphoreError' } | { type: 'UrlError' } | { type: 'TauriError' };
 
-export type LauncherEvent = { Ingress: IngressPayload } | { Message: MessagePayload } | { Process: ProcessPayload };
-
 export type MessageLevel = 'Info' | 'Warn' | 'Error';
 
 export interface MessagePayload { level: MessageLevel; message: string }
@@ -95,7 +93,7 @@ export interface Process { pid: number; started_at: string; cluster_id: bigint;
 
 export type ProcessError = { HookUnsuccessful: number } | 'NoPid';
 
-export type ProcessPayload = { Starting: { command: string } } | { Started: { process: Process } } | { Stopped: { pid: number; exit_code: number } } | { Output: { pid: number; output: string } };
+export type ProcessPayload = { type: 'Starting'; command: string } | { type: 'Started'; process: Process } | { type: 'Stopped'; pid: number; exit_code: number } | { type: 'Output'; pid: number; output: string };
 
 export interface Resolution { width: number; height: number }
 
@@ -103,11 +101,11 @@ export interface SettingProfileModel { name: string; java_id: bigint | null; res
 
 export interface SettingsOsExtra { enable_gamemode: boolean | null }
 
-const ARGS_MAP = { 'core': '{"openMsaLogin":[],"getClusters":[],"getClusterById":["id"],"getUsers":[],"removeUser":["uuid"],"removeCluster":["id"],"getGlobalProfile":[],"getUser":["uuid"],"setDefaultUser":["uuid"],"createCluster":["options"],"getDefaultUser":["fallback"],"getProfileOrDefault":["name"]}', 'oneclient': '{"return_error":[],"open_dev_tools":[]}', '': '{"send_event":["event"]}' };
+const ARGS_MAP = { oneclient: '{"open_dev_tools":[],"return_error":[]}', core: '{"getClusters":[],"createCluster":["options"],"getGlobalProfile":[],"getClusterById":["id"],"openMsaLogin":[],"getUsers":[],"getDefaultUser":["fallback"],"setDefaultUser":["uuid"],"getUser":["uuid"],"getProfileOrDefault":["name"],"removeUser":["uuid"],"removeCluster":["id"],"launchCluster":["id","uuid"]}', events: '{"message":["event"],"process":["event"],"ingress":["event"]}' };
 export interface Router {
-	'': { send_event: (event: LauncherEvent) => Promise<void> };
-	'core': { getClusters: () => Promise<Array<Model>>; getClusterById: (id: bigint) => Promise<Model | null>; removeCluster: (id: bigint) => Promise<null>; createCluster: (options: CreateCluster) => Promise<Model>; getProfileOrDefault: (name: string | null) => Promise<SettingProfileModel>; getGlobalProfile: () => Promise<SettingProfileModel>; getUsers: () => Promise<Array<MinecraftCredentials>>; getUser: (uuid: string) => Promise<MinecraftCredentials | null>; removeUser: (uuid: string) => Promise<null>; getDefaultUser: (fallback: boolean | null) => Promise<MinecraftCredentials | null>; setDefaultUser: (uuid: string | null) => Promise<null>; openMsaLogin: () => Promise<MinecraftCredentials | null> };
-	'oneclient': { return_error: () => Promise<null>; open_dev_tools: () => Promise<void> };
+	events: { ingress: (event: IngressPayload) => Promise<void>; message: (event: MessagePayload) => Promise<void>; process: (event: ProcessPayload) => Promise<void> };
+	core: { getClusters: () => Promise<Array<Model>>; getClusterById: (id: bigint) => Promise<Model | null>; removeCluster: (id: bigint) => Promise<null>; createCluster: (options: CreateCluster) => Promise<Model>; launchCluster: (id: bigint, uuid: string | null) => Promise<null>; getProfileOrDefault: (name: string | null) => Promise<SettingProfileModel>; getGlobalProfile: () => Promise<SettingProfileModel>; getUsers: () => Promise<Array<MinecraftCredentials>>; getUser: (uuid: string) => Promise<MinecraftCredentials | null>; removeUser: (uuid: string) => Promise<null>; getDefaultUser: (fallback: boolean | null) => Promise<MinecraftCredentials | null>; setDefaultUser: (uuid: string | null) => Promise<null>; openMsaLogin: () => Promise<MinecraftCredentials | null> };
+	oneclient: { return_error: () => Promise<null>; open_dev_tools: () => Promise<void> };
 }
 
 export type { InferCommandOutput };

apps/oneclient/frontend/src/routes/index.tsx

@@ -1,3 +1,4 @@
+import { bindings } from '@/main';
 import { Button } from '@onelauncher/common/components';
 import { createFileRoute } from '@tanstack/react-router';
 
@@ -6,10 +7,30 @@ export const Route = createFileRoute('/')({
 });
 
 function App() {
+	const onClick = async () => {
+		// const result = await bindings.core.createCluster({
+		// 	name: 'Test 1.8.9 Forge',
+		// 	mc_loader: 'forge',
+		// 	mc_version: '1.8.9',
+		// 	icon: null,
+		// 	mc_loader_version: '11.15.1.2318',
+		// });
+
+		await bindings.core.launchCluster(2 as unknown as bigint, null);
+
+		const unlisten = await bindings.events.process.on((e) => {
+			if (e.type === 'Output') { console.log(e.output); }
+			else if (e.type === 'Stopped') {
+				console.log(e.exit_code);
+				unlisten();
+			}
+		});
+	};
+
 	return (
 		<div className="text-center">
 			<h1>test</h1>
-			<Button>Click Me!</Button>
+			<Button onClick={onClick}>Click Me!</Button>
 		</div>
 	);
 }

packages/core/Cargo.toml

@@ -70,6 +70,7 @@ whoami = { workspace = true }
 rand = { workspace = true }
 url = { workspace = true }
 async-trait = { workspace = true }
+async-stream = { workspace = true }
 paste = { workspace = true }
 byteorder = { workspace = true }
 merge = { workspace = true }

packages/core/src/api/credentials.rs

@@ -84,3 +84,14 @@ pub async fn get_user(user: uuid::Uuid) -> LauncherResult<Option<MinecraftCreden
 
 	Ok(user)
 }
+
+/// Get a fake user for testing or offline mode.
+pub fn get_fake_user() -> MinecraftCredentials {
+	MinecraftCredentials {
+		id: uuid::Uuid::new_v4(),
+		username: "Player".to_string(),
+		access_token: "0".to_string(),
+		refresh_token: "0".to_string(),
+		expires: chrono::Utc::now() + chrono::Duration::days(1),
+	}
+}

packages/core/src/api/game/arguments.rs

@@ -3,7 +3,7 @@
 use std::collections::{HashMap, HashSet};
 use std::path::Path;
 
-use futures::AsyncReadExt;
+use futures::{pin_mut, AsyncReadExt, TryStreamExt};
 use interpulse::api::minecraft::{Argument, ArgumentValue, Library, VersionType};
 use interpulse::api::modded::SidedDataEntry;
 use interpulse::utils::get_path_from_artifact;
@@ -156,38 +156,48 @@ pub fn processor_arguments<T: AsRef<str>, S: ::std::hash::BuildHasher>(
 }
 
 pub async fn main_class(path: String) -> LauncherResult<Option<String>> {
-	// let zipfs = std::fs::File::open(&path).map_err(|e| IOError::with_path(e, &path))?;
-	// let mut archive = zip::ZipArchive::new(zipfs)
-	// 	.map_err(|_| anyhow::anyhow!("failed to read main class {}", path))?;
-	// let file = archive
-	// 	.by_name("META-INF/MANIFEST.MF")
-	// 	.map_err(|_| anyhow::anyhow!("failed to read manifest {}", path))?;
 	let data = io::read(path).await?;
 	let mut class_name: Option<String> = None;
+	let reader = io::stream_zip_entries_bytes(data);
 
-	io::read_zip_entries_bytes(data, async |_, entry, entry_reader| {
-		if entry.dir()? {
-			return Ok(());
+	pin_mut!(reader);
+	while let Ok(item) = reader.try_next().await {
+		let Some((index, entry, reader)) = item else {
+			continue;
+		};
+
+		if entry.dir().unwrap_or(true) {
+			continue;
 		}
 
-		if entry.filename().as_str()? != "META-INF/MANIFEST.MF" {
-			return Ok(());
+		if entry.filename().as_str().unwrap_or_default() != "META-INF/MANIFEST.MF" {
+			continue;
 		}
 
 		let mut buf = String::new();
-		entry_reader.read_to_string(&mut buf).await?;
+		let mut entry_reader = match reader.reader_without_entry(index).await {
+			Ok(reader) => reader,
+			Err(err) => {
+				tracing::error!("failed to get entry reader for {}: {}", entry.filename().as_str().unwrap_or_default(), err);
+				continue;
+			}
+		};
+
+		if let Err(err) = entry_reader.read_to_string(&mut buf).await {
+			tracing::error!("failed to read entry {}: {}", entry.filename().as_str().unwrap_or_default(), err);
+			continue;
+		}
 
 		for line in buf.lines() {
 			let line = line.trim();
-			if line.starts_with("Main-Class:")
-				&& let Some(class) = line.split(':').nth(1) {
-					class_name = Some(class.to_string());
-					return Ok(());
+			if line.starts_with("Main-Class:") {
+				if let Some(class) = line.split(':').nth(1) {
+					class_name = Some(class.trim().to_string());
+					break;
 				}
+			}
 		}
-
-		Ok(())
-	}).await?;
+	}
 
 	Ok(class_name)
 }

packages/core/src/api/game/launch.rs

@@ -30,7 +30,6 @@ pub async fn launch_minecraft(
 	}
 
 	prepare_cluster(cluster, force).await?;
-
 	let mut settings = get_global_profile().await;
 	if let Some(name) = &cluster.setting_profile_name
 		&& let Some(profile) = get_profile_by_name(name).await? {

packages/core/src/api/game/metadata.rs

@@ -41,9 +41,9 @@ pub async fn download_minecraft(
 
 	let asset_index = download_assets_index(version, sub_ingress, force).await?;
 	tokio::try_join! {
-		download_assets(version.assets == "legacy", &asset_index, sub_ingress, force),
+		download_assets(version.assets == "legacy", asset_index, sub_ingress, force),
 		download_client(version, sub_ingress, force),
-		download_libraries(&version.id, &version.libraries, java_arch, sub_ingress, force),
+		download_libraries(version.id.clone(), version.libraries.clone(), java_arch.to_string(), sub_ingress, force),
 	}?;
 
 	Ok(())
@@ -122,9 +122,6 @@ pub async fn download_assets_index(
 		.await?
 		.join(format!("{}.json", version.asset_index.id));
 
-	// let ingress_ref = ingress_ref.map(|i| i.with_increment(i.increment_by / 2.0));
-	// let ingress_ref = ingress_ref.as_ref();
-	// ingress_ref.map(async |i| send_ingress_ref_message(i, Some("fetching assets index")).await);
 	let ingress_step = ingress.ingress_total().map(|i| i / 2.0).unwrap_or_default();
 	ingress.set_ingress_message("fetching assets index").await?;
 
@@ -159,9 +156,9 @@ pub async fn download_assets_index(
 // MARK: Assets
 /// Downloads the assets for a given asset index, returns the number of failed downloads
 #[tracing::instrument(skip_all)]
-pub async fn download_assets(
+pub async fn download_assets<'a>(
 	legacy: bool,
-	assets_index: &AssetsIndex,
+	assets_index: AssetsIndex,
 	ingress: Option<&SubIngress<'_>>,
 	force: Option<bool>,
 ) -> LauncherResult<usize> {
@@ -178,30 +175,33 @@ pub async fn download_assets(
 
 	ingress.set_ingress_message("fetching assets").await?;
 
-	let requests = stream::iter(assets_index.objects.iter().map(|(name, asset)| async {
-		let hash = &asset.hash;
-		let subhash = &hash[0..2];
-		let path = if legacy {
-			dir.join(name.replace('/', std::path::MAIN_SEPARATOR_STR))
-		} else {
-			dir.join(subhash).join(hash)
-		};
+	let requests = stream::iter(assets_index.objects.into_iter().map(|(name, asset)| {
+		let dir = dir.clone();
+		async move {
+			let hash = &asset.hash;
+			let subhash = &hash[0..2];
+			let path = if legacy {
+				dir.join(name.replace('/', std::path::MAIN_SEPARATOR_STR))
+			} else {
+				dir.join(subhash).join(hash)
+			};
 
-		if !path.exists() || force.unwrap_or(false) {
-			http::download_advanced(
-				Method::GET,
-				&format!("https://resources.download.minecraft.net/{subhash}/{hash}"),
-				path,
-				None,
-				None,
-				Some((HashAlgorithm::Sha1, hash)),
-				ingress.ingress_sub(|total| total / len as f64).as_ref(),
-			)
-			.await
-			.map(|_| ())
-		} else {
-			// TODO: Possibly check hash? (not sure if this is a good idea here)
-			Ok(())
+			if !path.exists() || force.unwrap_or(false) {
+				http::download_advanced(
+					Method::GET,
+					&format!("https://resources.download.minecraft.net/{subhash}/{hash}"),
+					path,
+					None,
+					None,
+					Some((HashAlgorithm::Sha1, hash)),
+					ingress.ingress_sub(|total| total / len as f64).as_ref(),
+				)
+				.await
+				.map(|_| ())
+			} else {
+				// TODO: Possibly check hash? (not sure if this is a good idea here)
+				Ok(())
+			}
 		}
 	}))
 	.buffer_unordered(Core::get().fetch_attempts.min(7))
@@ -265,9 +265,9 @@ pub async fn download_client(
 // MARK: Libraries
 /// Downloads the libraries and returns the number of failed downloads
 pub async fn download_libraries(
-	version: &str,
-	libraries: &[Library],
-	java_arch: &str,
+	version: String,
+	libraries: Vec<Library>,
+	java_arch: String,
 	ingress: Option<&SubIngress<'_>>,
 	force: Option<bool>,
 ) -> LauncherResult<usize> {
@@ -284,88 +284,95 @@ pub async fn download_libraries(
 	let ingress = ingress.as_ref();
 	ingress.set_ingress_message("fetching libraries").await?;
 
-	let requests = stream::iter(libraries.iter().map(|lib| async {
-		if let Some(rules) = &lib.rules
-			&& !super::rules::validate_rules(rules, java_arch, lib.natives.is_some()) {
-				tracing::debug!("skipping library {} due to rules", lib.name);
-				return Ok::<(), LauncherError>(());
-			}
+	let requests = stream::iter(libraries.into_iter().map(|lib| {
+		let lib_dir = lib_dir.clone();
+		let natives_dest = natives_dest.clone();
+		let java_arch = java_arch.clone();
 
-		if !lib.downloadable {
-			tracing::debug!("skipping library {} due to downloadability", lib.name);
-			return Ok(());
-		}
-
-		let artifact_path = interpulse::utils::get_path_from_artifact(&lib.name)?;
-		let path = lib_dir.join(&artifact_path);
-
-		if path.exists() && !force.unwrap_or(false) {
-			tracing::debug!("library {} is installed, skipping", &lib.name);
-			return Ok(());
-		}
-
-		tokio::try_join! {
-			async {
-				if let Some(interpulse::api::minecraft::LibraryDownloads {
-					artifact: Some(ref artifact), ..
-				}) = lib.downloads
-					&& !artifact.url.is_empty() {
-						http::download(
-							Method::GET,
-							&artifact.url,
-							&path,
-							Some((HashAlgorithm::Sha1, &artifact.sha1)),
-							ingress,
-						).await?;
-
-						tracing::trace!("fetched library {} to path {:?}", &lib.name, &path);
-						return Ok::<_, LauncherError>(());
-					}
-
-				let url = [lib.url.as_deref().unwrap_or("https://libraries.minecraft.net/"), &artifact_path].concat();
-				http::download(
-					Method::GET,
-					&url,
-					&path,
-					None,
-					ingress,
-				).await?;
-
-				tracing::trace!("fetched library {} to path {:?}", &lib.name, &path);
-				Ok::<_, LauncherError>(())
-			},
-			async {
-				if let Some((os_key, classifiers)) = None.or_else(|| Some((
-					lib.natives.as_ref()?.get(&Os::native_arch(java_arch))?,
-					lib.downloads.as_ref()?.classifiers.as_ref()?
-				))) {
-					let parsed = os_key.replace("${arch}", crate::constants::ARCH_WIDTH);
-					if let Some(native) = classifiers.get(&parsed) {
-						tracing::trace!("found native library {}", &lib.name);
-
-						let data = http::fetch_advanced(
-							Method::GET,
-							&native.url,
-							None,
-							Some((HashAlgorithm::Sha1, &native.sha1)),
-							None,
-							ingress,
-						).await?;
-
-						io::unzip_bytes_filtered(
-							data.to_vec(),
-							Some(|name: &str| !name.starts_with("META-INF")),
-							&natives_dest
-						).await?;
-						tracing::trace!("extracted native {} to path {:?}", &lib.name, &natives_dest);
-					}
+		async move {
+			if let Some(rules) = &lib.rules
+				&& !super::rules::validate_rules(rules, &java_arch, lib.natives.is_some()) {
+					tracing::debug!("skipping library {} due to rules", lib.name);
+					return Ok::<(), LauncherError>(());
 				}
 
-				Ok(())
+			if !lib.downloadable {
+				tracing::debug!("skipping library {} due to downloadability", lib.name);
+				return Ok(());
 			}
-		}?;
 
-		Ok(())
+			let artifact_path = interpulse::utils::get_path_from_artifact(&lib.name)?;
+			let path = lib_dir.join(&artifact_path);
+
+			if path.exists() && !force.unwrap_or(false) {
+				tracing::debug!("library {} is installed, skipping", &lib.name);
+				return Ok(());
+			}
+
+			tokio::try_join! {
+				async {
+					if let Some(interpulse::api::minecraft::LibraryDownloads {
+						artifact: Some(ref artifact), ..
+					}) = lib.downloads
+						&& !artifact.url.is_empty() {
+							http::download(
+								Method::GET,
+								&artifact.url,
+								&path,
+								Some((HashAlgorithm::Sha1, &artifact.sha1)),
+								ingress,
+							).await?;
+
+							tracing::trace!("fetched library {} to path {:?}", &lib.name, &path);
+							return Ok::<_, LauncherError>(());
+						}
+
+					let url = [lib.url.as_deref().unwrap_or("https://libraries.minecraft.net/"), &artifact_path].concat();
+					http::download(
+						Method::GET,
+						&url,
+						&path,
+						None,
+						ingress,
+					).await?;
+
+					tracing::trace!("fetched library {} to path {:?}", &lib.name, &path);
+					Ok::<_, LauncherError>(())
+				},
+				async {
+					if let Some((os_key, classifiers)) = None.or_else(|| Some((
+						lib.natives.as_ref()?.get(&Os::native_arch(&java_arch))?,
+						lib.downloads.as_ref()?.classifiers.as_ref()?
+					))) {
+						let parsed = os_key.replace("${arch}", crate::constants::ARCH_WIDTH);
+						if let Some(native) = classifiers.get(&parsed) {
+							tracing::trace!("found native library {}", &lib.name);
+
+							let data = http::fetch_advanced(
+								Method::GET,
+								&native.url,
+								None,
+								Some((HashAlgorithm::Sha1, &native.sha1)),
+								None,
+								ingress,
+							).await?;
+
+							io::unzip_bytes_filtered(
+								data.to_vec(),
+								Some(|name: &str| !name.starts_with("META-INF")),
+								&natives_dest
+							).await?;
+
+							tracing::trace!("extracted native {} to path {:?}", &lib.name, &natives_dest);
+						}
+					}
+
+					Ok(())
+				}
+			}?;
+
+			Ok(())
+		}
 	})).buffer_unordered(7).collect::<Vec<_>>();
 
 	let mut failed = 0;

packages/core/src/api/processes.rs

@@ -4,6 +4,7 @@ use crate::store::processes::Process;
 
 #[onelauncher_macro::specta]
 #[derive(Debug, Clone, Serialize)]
+#[serde(tag = "type")]
 pub enum ProcessPayload {
 	Starting {
 		command: String,

packages/core/src/api/proxy/impl/proxy_tauri.rs

@@ -30,7 +30,11 @@ impl LauncherProxy for ProxyTauri {
 			}
 		}
 
-		Ok(self.emitter.send_event(event)?)
+		Ok(match event {
+			LauncherEvent::Ingress(ingress) => self.emitter.ingress(ingress),
+			LauncherEvent::Message(message) => self.emitter.message(message),
+			LauncherEvent::Process(process) => self.emitter.process(process),
+		}?)
 	}
 
 	#[tracing::instrument]

packages/core/src/api/tauri/commands.rs

@@ -21,6 +21,10 @@ pub trait TauriLauncherApi {
 	#[taurpc(alias = "createCluster")]
 	async fn create_cluster(options: CreateCluster) -> LauncherResult<Cluster>;
 
+	#[taurpc(alias = "launchCluster")]
+	async fn launch_cluster(id: ClusterId, uuid: Option<uuid::Uuid>) -> LauncherResult<()>;
+
+
 	// Setting Profiles
 	#[taurpc(alias = "getProfileOrDefault")]
 	async fn get_profile_or_default(name: Option<String>) -> LauncherResult<SettingsProfile>;
@@ -91,6 +95,25 @@ impl TauriLauncherApi for TauriLauncherApiImpl {
 		Ok(cluster)
 	}
 
+	async fn launch_cluster(self, id: ClusterId, uuid: Option<uuid::Uuid>) -> LauncherResult<()> {
+		let mut cluster = api::cluster::dao::get_cluster_by_id(id).await?
+			.ok_or_else(|| anyhow::anyhow!("cluster with id {} not found", id))?;
+
+		let uuid = match uuid {
+			Some(uuid) => uuid,
+			None => api::credentials::get_default_user().await?.ok_or_else(|| anyhow::anyhow!("no default user set"))?,
+		};
+
+		let user = api::credentials::get_user(uuid).await?
+			.ok_or_else(|| anyhow::anyhow!("user with uuid {} not found", uuid))?;
+
+		// let user = api::credentials::get_fake_user();
+
+		let _ = api::game::launch::launch_minecraft(&mut cluster, user, None).await?;
+
+		Ok(())
+	}
+
 	async fn get_profile_or_default(self, name: Option<String>) -> LauncherResult<SettingsProfile> {
 		api::setting_profiles::dao::get_profile_or_default(name.as_ref()).await
 	}
@@ -186,4 +209,3 @@ impl TauriLauncherApi for TauriLauncherApiImpl {
 	}
 
 }
-

packages/core/src/api/tauri/events.rs

@@ -1,9 +1,16 @@
-use crate::api::proxy::event::LauncherEvent;
-
-#[taurpc::procedures(event_trigger = LauncherEventEmitter)]
+#[taurpc::procedures(path = "events", event_trigger = LauncherEventEmitter)]
 pub trait TauriLauncherEventApi {
 	#[taurpc(event)]
-	async fn send_event(event: LauncherEvent);
+	#[taurpc(alias = "ingress")]
+	async fn ingress(event: crate::api::ingress::IngressPayload);
+
+	#[taurpc(event)]
+	#[taurpc(alias = "message")]
+	async fn message(event: crate::api::proxy::message::MessagePayload);
+
+	#[taurpc(event)]
+	#[taurpc(alias = "process")]
+	async fn process(event: crate::api::processes::ProcessPayload);
 }
 
 #[taurpc::ipc_type]

packages/core/src/lib.rs

@@ -30,4 +30,4 @@ pub async fn initialize_core(options: CoreOptions, proxy_backend: impl LauncherP
 	tracing::info!("Core initialized successfully");
 
 	Ok(())
-}
+}

packages/core/src/utils/crypto.rs

@@ -32,76 +32,54 @@ pub enum HashAlgorithm {
 const BUFFER_SIZE: usize = 1024 * 256; // 16 KiB
 
 impl HashAlgorithm {
-	pub fn hasher_sync<F>(&self, closure: F) -> Result<String, IOError>
-	where
-		F: Fn(&mut (dyn DynDigest)) -> Result<(), IOError>,
-	{
-		let mut hasher = get_inner(self);
-		closure(&mut *hasher)?;
-		Ok(to_hex(&hasher.finalize()))
-	}
-
-	pub async fn hasher<F>(&self, closure: F) -> Result<String, IOError>
-	where
-		F: AsyncFn(&mut (dyn DynDigest + Send + Sync)) -> Result<(), IOError> + Send + Sync,
-	{
-		let mut hasher = get_inner(self);
-		closure(&mut *hasher).await?;
-		Ok(to_hex(&hasher.finalize()))
-	}
-
 	pub async fn hash(&self, data: &[u8]) -> Result<String, IOError> {
-		self.hasher(async move |hasher| {
-			hasher.update(data);
-			Ok(())
-		}).await
+		let mut hasher = get_inner(self);
+		hasher.update(data);
+		Ok(to_hex(&hasher.finalize()))
 	}
 
 	pub fn hash_sync(&self, data: &[u8]) -> Result<String, IOError> {
-		self.hasher_sync(move |hasher| {
-			hasher.update(data);
-			Ok(())
-		})
+		let mut hasher = get_inner(self);
+		hasher.update(data);
+		Ok(to_hex(&hasher.finalize()))
 	}
 
 	pub async fn hash_file(&self, path: impl AsRef<std::path::Path>) -> Result<String, IOError> {
 		let path = path.as_ref();
-		self.hasher(async |hasher: &mut (dyn DynDigest + Send + Sync)| {
-			let file = tokio::fs::File::open(path).await?;
-			let mut reader = tokio::io::BufReader::new(file);
 
-			let mut buffer = vec![0; BUFFER_SIZE].into_boxed_slice();
-			loop {
-				let bytes_read = reader.read(&mut buffer).await?;
-				if bytes_read == 0 {
-					break;
-				}
+		let mut hasher = get_inner(self);
 
-				hasher.update(&buffer[..bytes_read]);
+		let file = tokio::fs::File::open(path).await?;
+		let mut reader = tokio::io::BufReader::new(file);
+
+		let mut buffer = vec![0; BUFFER_SIZE].into_boxed_slice();
+		loop {
+			let bytes_read = reader.read(&mut buffer).await?;
+			if bytes_read == 0 {
+				break;
 			}
 
-			Ok(())
-		}).await
+			hasher.update(&buffer[..bytes_read]);
+		}
+
+		Ok(to_hex(&hasher.finalize()))
 	}
 
 	pub fn hash_file_sync(&self, path: impl AsRef<std::path::Path>) -> Result<String, IOError> {
-		self.hasher_sync(|hasher: &mut dyn DynDigest| {
-			let path = path.as_ref();
-			let file = std::fs::File::open(path)?;
-			let mut reader = std::io::BufReader::new(file);
+		let path = path.as_ref();
+		let mut hasher = get_inner(self);
+		let mut file = std::fs::File::open(path)?;
 
-			let mut buffer = vec![0; BUFFER_SIZE].into_boxed_slice();
-			loop {
-				let bytes_read = reader.read(&mut buffer)?;
-				if bytes_read == 0 {
-					break;
-				}
-
-				hasher.update(&buffer[..bytes_read]);
+		let mut buffer = vec![0; BUFFER_SIZE].into_boxed_slice();
+		loop {
+			let bytes_read = file.read(&mut buffer)?;
+			if bytes_read == 0 {
+				break;
 			}
 
-			Ok(())
-		})
+			hasher.update(&buffer[..bytes_read]);
+		}
+		Ok(to_hex(&hasher.finalize()))
 	}
 }
 

packages/core/src/utils/io.rs

@@ -1,8 +1,10 @@
-use std::path::PathBuf;
+use std::{path::PathBuf, sync::Arc};
 
 use async_compression::tokio::bufread::GzipDecoder;
+use async_stream::try_stream;
 use async_zip::StoredZipEntry;
 use async_zip::base::read::WithoutEntry;
+use futures::{pin_mut, Stream, TryStreamExt};
 use serde::Serialize;
 use serde::de::DeserializeOwned;
 use tempfile::TempDir;
@@ -278,6 +280,24 @@ where
 	Ok(())
 }
 
+/// Reads a zip archive from a byte array and returns a stream of entries.
+pub fn stream_zip_entries_bytes(
+	data: Vec<u8>,
+) -> impl Stream<Item = Result<(
+	usize, StoredZipEntry, Arc<async_zip::base::read::mem::ZipFileReader>
+), IOError>> {
+	try_stream! {
+		let reader = Arc::new(async_zip::base::read::mem::ZipFileReader::new(data).await?);
+		let entries = reader.file().entries();
+
+		for index in 0..entries.len() {
+			let entry = entries.get(index).expect("expected more zip entries");
+
+			yield (index, entry.clone(), reader.clone());
+		}
+	}
+}
+
 /// Unzips a zip archive from a byte array
 pub async fn unzip_bytes(
 	data: Vec<u8>,
@@ -289,15 +309,18 @@ pub async fn unzip_bytes(
 /// Unzips a zip archive from a byte array
 pub async fn unzip_bytes_filtered(
 	data: Vec<u8>,
-	filter_entries: Option<impl Fn(&str) -> bool>,
+	filter_entries: Option<impl Fn(&str) -> bool + Send + Sync>,
 	dest_path: impl AsRef<std::path::Path>,
 ) -> Result<(), IOError> {
-	read_zip_entries_bytes(data, async |_, entry, entry_reader| {
+	let stream = stream_zip_entries_bytes(data);
+	pin_mut!(stream);
+
+	while let Some((index, entry, reader)) = stream.try_next().await? {
 		let file_name = entry.filename().as_str()?;
 
 		if let Some(filter) = &filter_entries
 			&& !filter(file_name) {
-				return Ok(());
+				continue;
 			}
 
 		let path = dest_path.as_ref().join(sanitize_path(file_name));
@@ -311,15 +334,14 @@ pub async fn unzip_bytes_filtered(
 				create_dir_all(parent).await?;
 			}
 
+			let entry_reader = reader.reader_without_entry(index).await?;
+
 			let file = tokio::fs::File::create(&path).await?;
 			let writer = tokio::io::BufWriter::new(file);
 
 			futures_lite::io::copy(entry_reader, &mut writer.compat_write()).await?;
 		}
-
-		Ok(())
-	})
-	.await?;
+	}
 
 	Ok(())
 }