From 17734ca1b9033a35ee6a581cc55e25fa7fc932b3 Mon Sep 17 00:00:00 2001 From: LynithDev <61880709+LynithDev@users.noreply.github.com> Date: Sat, 7 Jun 2025 11:51:46 +0200 Subject: [PATCH] fix: made functions sendable between threads --- Cargo.lock | 1 + Cargo.toml | 1 + apps/oneclient/frontend/src/bindings.gen.ts | 12 +- apps/oneclient/frontend/src/routes/index.tsx | 23 +- packages/core/Cargo.toml | 1 + packages/core/src/api/credentials.rs | 11 + packages/core/src/api/game/arguments.rs | 50 ++-- packages/core/src/api/game/launch.rs | 1 - packages/core/src/api/game/metadata.rs | 227 +++++++++--------- packages/core/src/api/processes.rs | 1 + .../core/src/api/proxy/impl/proxy_tauri.rs | 6 +- packages/core/src/api/tauri/commands.rs | 24 +- packages/core/src/api/tauri/events.rs | 15 +- packages/core/src/lib.rs | 2 +- packages/core/src/utils/crypto.rs | 82 +++---- packages/core/src/utils/io.rs | 38 ++- 16 files changed, 289 insertions(+), 206 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6727a4ff..34bbd097 100644 --- a/Cargo.lock +++ b/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", diff --git a/Cargo.toml b/Cargo.toml index c2860f37..033bcae0 100644 --- a/Cargo.toml +++ b/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" } diff --git a/apps/oneclient/frontend/src/bindings.gen.ts b/apps/oneclient/frontend/src/bindings.gen.ts index 88ad8162..ad95f510 100644 --- a/apps/oneclient/frontend/src/bindings.gen.ts +++ b/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 }; - 'core': { getClusters: () => Promise>; getClusterById: (id: bigint) => Promise; removeCluster: (id: bigint) => Promise; createCluster: (options: CreateCluster) => Promise; getProfileOrDefault: (name: string | null) => Promise; getGlobalProfile: () => Promise; getUsers: () => Promise>; getUser: (uuid: string) => Promise; removeUser: (uuid: string) => Promise; getDefaultUser: (fallback: boolean | null) => Promise; setDefaultUser: (uuid: string | null) => Promise; openMsaLogin: () => Promise }; - 'oneclient': { return_error: () => Promise; open_dev_tools: () => Promise }; + events: { ingress: (event: IngressPayload) => Promise; message: (event: MessagePayload) => Promise; process: (event: ProcessPayload) => Promise }; + core: { getClusters: () => Promise>; getClusterById: (id: bigint) => Promise; removeCluster: (id: bigint) => Promise; createCluster: (options: CreateCluster) => Promise; launchCluster: (id: bigint, uuid: string | null) => Promise; getProfileOrDefault: (name: string | null) => Promise; getGlobalProfile: () => Promise; getUsers: () => Promise>; getUser: (uuid: string) => Promise; removeUser: (uuid: string) => Promise; getDefaultUser: (fallback: boolean | null) => Promise; setDefaultUser: (uuid: string | null) => Promise; openMsaLogin: () => Promise }; + oneclient: { return_error: () => Promise; open_dev_tools: () => Promise }; } export type { InferCommandOutput }; diff --git a/apps/oneclient/frontend/src/routes/index.tsx b/apps/oneclient/frontend/src/routes/index.tsx index 06960add..7f4a4afa 100644 --- a/apps/oneclient/frontend/src/routes/index.tsx +++ b/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 (

test

- +
); } diff --git a/packages/core/Cargo.toml b/packages/core/Cargo.toml index 2a63013e..fba93f28 100644 --- a/packages/core/Cargo.toml +++ b/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 } diff --git a/packages/core/src/api/credentials.rs b/packages/core/src/api/credentials.rs index e2229123..733d639b 100644 --- a/packages/core/src/api/credentials.rs +++ b/packages/core/src/api/credentials.rs @@ -84,3 +84,14 @@ pub async fn get_user(user: uuid::Uuid) -> LauncherResult 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), + } +} \ No newline at end of file diff --git a/packages/core/src/api/game/arguments.rs b/packages/core/src/api/game/arguments.rs index c3c30264..5a75a74a 100644 --- a/packages/core/src/api/game/arguments.rs +++ b/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, S: ::std::hash::BuildHasher>( } pub async fn main_class(path: String) -> LauncherResult> { - // 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 = 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) } diff --git a/packages/core/src/api/game/launch.rs b/packages/core/src/api/game/launch.rs index 7aa07f7f..9771d4e6 100644 --- a/packages/core/src/api/game/launch.rs +++ b/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? { diff --git a/packages/core/src/api/game/metadata.rs b/packages/core/src/api/game/metadata.rs index 5dc95cdf..23e38502 100644 --- a/packages/core/src/api/game/metadata.rs +++ b/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, ) -> LauncherResult { @@ -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, + java_arch: String, ingress: Option<&SubIngress<'_>>, force: Option, ) -> LauncherResult { @@ -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::>(); let mut failed = 0; diff --git a/packages/core/src/api/processes.rs b/packages/core/src/api/processes.rs index ef21db3c..c4ecccf4 100644 --- a/packages/core/src/api/processes.rs +++ b/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, diff --git a/packages/core/src/api/proxy/impl/proxy_tauri.rs b/packages/core/src/api/proxy/impl/proxy_tauri.rs index edd8b7e7..f38d0558 100644 --- a/packages/core/src/api/proxy/impl/proxy_tauri.rs +++ b/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] diff --git a/packages/core/src/api/tauri/commands.rs b/packages/core/src/api/tauri/commands.rs index 8e8ef1a3..770563e6 100644 --- a/packages/core/src/api/tauri/commands.rs +++ b/packages/core/src/api/tauri/commands.rs @@ -21,6 +21,10 @@ pub trait TauriLauncherApi { #[taurpc(alias = "createCluster")] async fn create_cluster(options: CreateCluster) -> LauncherResult; + #[taurpc(alias = "launchCluster")] + async fn launch_cluster(id: ClusterId, uuid: Option) -> LauncherResult<()>; + + // Setting Profiles #[taurpc(alias = "getProfileOrDefault")] async fn get_profile_or_default(name: Option) -> LauncherResult; @@ -91,6 +95,25 @@ impl TauriLauncherApi for TauriLauncherApiImpl { Ok(cluster) } + async fn launch_cluster(self, id: ClusterId, uuid: Option) -> 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) -> LauncherResult { api::setting_profiles::dao::get_profile_or_default(name.as_ref()).await } @@ -186,4 +209,3 @@ impl TauriLauncherApi for TauriLauncherApiImpl { } } - diff --git a/packages/core/src/api/tauri/events.rs b/packages/core/src/api/tauri/events.rs index 6ff19fab..26170a47 100644 --- a/packages/core/src/api/tauri/events.rs +++ b/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] diff --git a/packages/core/src/lib.rs b/packages/core/src/lib.rs index 5b64e6cb..df088b52 100644 --- a/packages/core/src/lib.rs +++ b/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(()) -} +} \ No newline at end of file diff --git a/packages/core/src/utils/crypto.rs b/packages/core/src/utils/crypto.rs index 8020dab4..1daa5134 100644 --- a/packages/core/src/utils/crypto.rs +++ b/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(&self, closure: F) -> Result - 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(&self, closure: F) -> Result - 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 { - 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 { - 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) -> Result { 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) -> Result { - 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())) } } diff --git a/packages/core/src/utils/io.rs b/packages/core/src/utils/io.rs index 1342b91f..cb376e2d 100644 --- a/packages/core/src/utils/io.rs +++ b/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, +) -> impl Stream +), 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, @@ -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, - filter_entries: Option bool>, + filter_entries: Option bool + Send + Sync>, dest_path: impl AsRef, ) -> 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(()) }