Ryan/drop-app
1
0
Fork 0
mirror of https://github.com/Drop-OSS/drop-app.git synced 2026-07-24 17:03:14 +10:00
Code Issues Packages Projects Releases Wiki Activity

Async downloader, better Proton support (#183)

Browse Source 
* feat: async downloader + other fixes

* feat: windows command parsing + use library path for install path

* feat: better proton support

* feat: style fixes and store button now uses in-app

* feat: emulator rename + umu emulator fix

* feat: bring process creation inline with docs

* fix: clippy
This commit is contained in:
DecDuck
2026-02-06 23:24:14 +11:00
committed by GitHub
parent b71809c041
commit 01335dadaf
45 changed files with 1453 additions and 381 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/games/src/downloads/download_agent.rs
+117 -122
View File
@@ -11,15 +11,17 @@ use download_manager::util::download_thread_control_flag::{
};
use download_manager::util::progress_object::{ProgressHandle, ProgressObject, ProgressType};
use droplet_rs::manifest::{ChunkData, Manifest};
use futures_util::StreamExt;
use futures_util::stream::FuturesUnordered;
use log::{debug, error, info, warn};
use remote::auth::generate_authorization_header;
use remote::cache::get_cached_object;
use remote::error::RemoteAccessError;
use remote::requests::generate_url;
use remote::utils::DROP_CLIENT_ASYNC;
use serde::Deserialize;
use std::collections::HashMap;
use std::fmt::Debug;
use std::mem;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Instant;
@@ -28,7 +30,7 @@ use tokio::sync::mpsc::Sender;
use utils::{app_emit, lock, send};
use crate::downloads::utils::get_disk_available;
use crate::library::{on_game_complete, push_game_update, set_partially_installed};
use crate::library::{Game, on_game_complete, push_game_update, set_partially_installed};
use crate::state::GameStatusManager;
use super::download_logic::download_game_chunk;
@@ -87,9 +89,11 @@ impl GameDownloadAgent {
// Don't run by default
let control_flag = DownloadThreadControl::new(DownloadThreadControlFlag::Stop);
let game_name = get_cached_object::<Game>(&format!("game/{}", metadata.id)).map(|v| v.library_path).unwrap_or(metadata.id.clone());
let base_dir_path = Path::new(&base_dir);
info!("base dir {}", base_dir_path.display());
let data_base_dir_path = base_dir_path.join(metadata.id.clone());
let data_base_dir_path = base_dir_path.join(game_name);
info!("data dir path {}", data_base_dir_path.display());
let stored_manifest = DropData::generate(
@@ -166,10 +170,7 @@ impl GameDownloadAgent {
info!("beginning download for {}...", self.metadata().id);
let res = self
.run()
.await
.map_err(ApplicationDownloadError::Communication);
let res = self.run().await;
debug!(
"{} took {}ms to download",
@@ -244,17 +245,16 @@ impl GameDownloadAgent {
self.download_progress
.set_max(dl_info.download_size.try_into().unwrap());
self.download_progress
.set_size(total_chunks);
self.download_progress.set_size(total_chunks);
self.download_progress.reset();
self.disk_progress.set_max(dl_info.install_size.try_into().unwrap());
self.disk_progress
.set_size(total_chunks);
.set_max(dl_info.install_size.try_into().unwrap());
self.disk_progress.set_size(total_chunks);
self.disk_progress.reset();
}
async fn run(&self) -> Result<bool, RemoteAccessError> {
async fn run(&self) -> Result<bool, ApplicationDownloadError> {
self.depot_manager.sync_depots().await?;
info!("synced depots");
self.setup_progress();
@@ -278,124 +278,119 @@ impl GameDownloadAgent {
completed_chunks.clone()
};
let chunk_len = manifests_chunks.iter().map(|v| v.1.len()).sum::<usize>();
let max_download_threads = borrow_db_checked().settings.max_download_threads;
let mut max_download_threads = borrow_db_checked().settings.max_download_threads;
if max_download_threads <= 0 {
max_download_threads = 1;
}
let (sender, recv) = crossbeam_channel::bounded(16);
// SAFETY: I pinky-promise
// (the scope keeps these in scope)
let unsafe_self: &'static GameDownloadAgent = unsafe { mem::transmute(self) };
let file_list: &'static HashMap<String, String> = unsafe { mem::transmute(&file_list) };
let file_list = &file_list;
let local_completed_chunks = completed_chunks.clone();
let download_join_handle = tauri::async_runtime::spawn_blocking(move || {
let thread_pool = rayon::ThreadPoolBuilder::new()
.num_threads(max_download_threads)
.build()
.unwrap();
thread_pool.scope(move |s| {
let mut index = 0;
for (version_id, chunks, key) in manifests_chunks.into_iter() {
let version_id = &version_id;
for (chunk_id, chunk_data) in chunks.into_iter() {
let local_sender = sender.clone();
let download_progress_handle = ProgressHandle::new(
unsafe_self.download_progress.get(index),
unsafe_self.download_progress.clone(),
);
let disk_progress_handle = ProgressHandle::new(
unsafe_self.disk_progress.get(index),
unsafe_self.disk_progress.clone(),
);
index += 1;
let chunk_length = chunk_data.files.iter().map(|v| v.length).sum();
if *local_completed_chunks.get(&chunk_id).unwrap_or(&false) {
download_progress_handle.skip(chunk_length);
continue;
}
let sender = unsafe_self.sender.clone();
let (depot, permit) = match unsafe_self
.depot_manager
.next_depot(&unsafe_self.metadata.id, &unsafe_self.metadata.version)
{
Ok(v) => v,
Err(err) => {
tauri::async_runtime::spawn(async move {
send!(
sender,
DownloadManagerSignal::Error(
ApplicationDownloadError::Communication(err)
)
);
});
return;
}
};
let local_version_id = version_id.clone();
s.spawn(move |_| {
for i in 0..RETRY_COUNT {
let base_path = unsafe_self.dropdata.base_path.clone();
match download_game_chunk(
&unsafe_self.metadata.id,
&local_version_id,
&chunk_id,
&depot,
&key,
&chunk_data,
file_list,
base_path,
&unsafe_self.control_flag,
&download_progress_handle,
&disk_progress_handle,
) {
Ok(true) => {
local_sender.send(chunk_id.clone()).unwrap();
drop(permit); // Take ownership
return;
}
Ok(false) => return,
Err(e) => {
warn!("got error for chunk id {}: {e:?}", chunk_id);
let retry = true; /*matches!(
&e,
ApplicationDownloadError::Communication(_)
| ApplicationDownloadError::Checksum
| ApplicationDownloadError::Lock
| ApplicationDownloadError::IoError(_)
);*/
if i == RETRY_COUNT - 1 || !retry {
warn!("retry logic failed, not re-attempting.");
tauri::async_runtime::spawn(async move {
send!(sender, DownloadManagerSignal::Error(e));
});
return;
}
}
}
}
});
}
}
drop(sender);
});
});
let mut chunk_completions = FuturesUnordered::new();
let mut outputs = Vec::new();
while let Ok(chunk_id) = recv.recv() {
outputs.push(chunk_id);
let mut handle_output =
|value: Result<Option<String>, ApplicationDownloadError>| match value {
Ok(value) => {
if let Some(chunk_id) = value {
outputs.push(chunk_id);
}
Ok(())
}
Err(err) => return Err(err),
};
let mut index = 0;
for (version_id, chunks, key) in manifests_chunks.into_iter() {
let version_id = &version_id;
for (chunk_id, chunk_data) in chunks.into_iter() {
let download_progress_handle = ProgressHandle::new(
self.download_progress.get(index),
self.download_progress.clone(),
);
let disk_progress_handle = ProgressHandle::new(
self.disk_progress.get(index),
self.disk_progress.clone(),
);
index += 1;
let chunk_length = chunk_data.files.iter().map(|v| v.length).sum();
if *local_completed_chunks.get(&chunk_id).unwrap_or(&false) {
download_progress_handle.skip(chunk_length);
continue;
}
let (depot, permit) = match self
.depot_manager
.next_depot(&self.metadata.id, &self.metadata.version)
{
Ok(v) => v,
Err(err) => {
return Err(err.into());
}
};
let local_version_id = version_id.clone();
while chunk_completions.len() >= max_download_threads {
handle_output(
chunk_completions
.next()
.await
.expect("max download threads is zero?"),
)?;
}
chunk_completions.push(async move {
for i in 0..RETRY_COUNT {
let base_path = self.dropdata.base_path.clone();
match download_game_chunk(
&self.metadata.id,
&local_version_id,
&chunk_id,
&depot,
&key,
&chunk_data,
file_list,
base_path,
&self.control_flag,
&download_progress_handle,
&disk_progress_handle,
)
.await
{
Ok(true) => {
drop(permit);
return Ok(Some(chunk_id.clone()));
}
Ok(false) => return Ok(None),
Err(e) => {
warn!("got error for chunk id {}: {e:?}", chunk_id);
let retry = true; /*matches!(
&e,
ApplicationDownloadError::Communication(_)
| ApplicationDownloadError::Checksum
| ApplicationDownloadError::Lock
| ApplicationDownloadError::IoError(_)
);*/
if i == RETRY_COUNT - 1 || !retry {
warn!("retry logic failed, not re-attempting.");
return Err(e);
}
}
}
}
Ok(None)
});
}
}
download_join_handle
.await
.expect("failed to complete download");
while let Some(value) = chunk_completions.next().await {
handle_output(value)?
}
for completed_chunk in outputs {
completed_chunks.insert(completed_chunk, true);
src-tauri/games/src/downloads/download_logic.rs
+20 -15
View File
@@ -1,6 +1,6 @@
use std::collections::HashMap;
use std::fs::{Permissions, set_permissions};
use std::io::{Read, Seek as _, SeekFrom, Write as _};
use std::io::SeekFrom;
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use std::path::PathBuf;
@@ -14,19 +14,22 @@ use download_manager::util::download_thread_control_flag::{
};
use download_manager::util::progress_object::ProgressHandle;
use droplet_rs::manifest::ChunkData;
use futures_util::StreamExt as _;
use log::{debug, info};
use remote::auth::generate_authorization_header;
use remote::error::{DropServerError, RemoteAccessError};
use remote::utils::DROP_CLIENT_SYNC;
use remote::utils::DROP_CLIENT_ASYNC;
use sha2::Digest;
use tauri::Url;
use tokio::io::{AsyncReadExt as _, AsyncSeekExt as _, AsyncWriteExt as _};
use tokio_util::io::StreamReader;
const READ_BUF_LEN: usize = 1024 * 1024;
type Aes128Ctr64LE = ctr::Ctr64LE<aes::Aes128>;
#[allow(clippy::too_many_arguments)]
pub fn download_game_chunk(
pub async fn download_game_chunk(
game_id: &str,
version_id: &str,
chunk_id: &str,
@@ -57,15 +60,16 @@ pub fn download_game_chunk(
.join(&format!("content/{}/{}/{}", game_id, version_id, chunk_id))
.map_err(|v| ApplicationDownloadError::DownloadError(v.into()))?;
let response = DROP_CLIENT_SYNC
let response = DROP_CLIENT_ASYNC
.get(url)
.header("Authorization", header)
.send()
.await
.map_err(|e| ApplicationDownloadError::Communication(e.into()))?;
if response.status() != 200 {
info!("chunk request got status code: {}", response.status());
let raw_res = response.text().map_err(|e| {
let raw_res = response.text().await.map_err(|e| {
ApplicationDownloadError::Communication(RemoteAccessError::FetchErrorLegacy(e.into()))
})?;
info!("{raw_res}");
@@ -89,11 +93,11 @@ pub fn download_game_chunk(
debug!("took {}ms to start downloading", timestep);
/*let stream = response
let stream = response
.bytes_stream()
.map(|v| v.map_err(|err| std::io::Error::other(err)));
let mut stream_reader = StreamReader::new(stream);*/
let mut stream_reader = response;
let mut stream_reader = StreamReader::new(stream);
//let mut stream_reader = response;
let mut hasher = sha2::Sha256::new();
let mut cipher = Aes128Ctr64LE::new(key.into(), &chunk_data.iv.into());
@@ -108,13 +112,14 @@ pub fn download_game_chunk(
std::fs::create_dir_all(parent)?;
}
let mut file_handle = if should_write {
let mut file_handle = std::fs::OpenOptions::new()
let mut file_handle = tokio::fs::OpenOptions::new()
.truncate(false)
.write(true)
.append(false)
.create(true)
.open(&path)?;
file_handle.seek(SeekFrom::Start(file.start.try_into().unwrap()))?;
.open(&path)
.await?;
file_handle.seek(SeekFrom::Start(file.start.try_into().unwrap())).await?;
Some(file_handle)
} else {
None
@@ -122,14 +127,14 @@ pub fn download_game_chunk(
let mut remaining = file.length;
while remaining > 0 {
let amount = stream_reader.read(&mut read_buf[0..remaining.min(READ_BUF_LEN)])?;
let amount = stream_reader.read(&mut read_buf[0..remaining.min(READ_BUF_LEN)]).await?;
download_progress.add(amount);
remaining -= amount;
cipher.apply_keystream(&mut read_buf[0..amount]);
//hasher.update(&read_buf[0..amount]);
hasher.update(&read_buf[0..amount]);
if let Some(file_handle) = &mut file_handle {
file_handle.write_all(&read_buf[0..amount])?;
file_handle.write_all(&read_buf[0..amount]).await?;
disk_progress.add(amount);
}
}
@@ -155,7 +160,7 @@ pub fn download_game_chunk(
let digest = hex::encode(hasher.finalize());
if digest != chunk_data.checksum {
//return Err(ApplicationDownloadError::Checksum);
return Err(ApplicationDownloadError::Checksum);
}
Ok(true)
src-tauri/games/src/library.rs
+7 -7
View File
@@ -49,6 +49,7 @@ pub struct Game {
pub m_cover_object_id: String,
pub m_image_library_object_ids: Vec<String>,
pub m_image_carousel_object_ids: Vec<String>,
pub library_path: String,
}
impl Game {
pub fn id(&self) -> &String {
@@ -246,6 +247,10 @@ pub async fn on_game_complete(
.applications
.game_statuses
.insert(meta.id.clone(), status.clone());
db_handle
.applications
.transient_statuses
.remove(meta);
drop(db_handle);
app_emit!(
app_handle,
@@ -289,11 +294,6 @@ pub fn push_game_update(
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct FrontendGameOptions {
launch_string: String,
}
impl FrontendGameOptions {
pub fn launch_string(&self) -> &String {
&self.launch_string
}
pub launch_string: String,
pub override_proton_path: Option<String>,
}
src-tauri/games/src/scan.rs
+1 -2
View File
@@ -19,7 +19,6 @@ pub fn scan_install_dirs() {
if !drop_data_file.exists() {
continue;
}
let game_id = game.file_name().display().to_string();
let Ok(drop_data) = DropData::read(&game.path()) else {
warn!(
".dropdata exists for {}, but couldn't read it. is it corrupted?",
@@ -27,7 +26,7 @@ pub fn scan_install_dirs() {
);
continue;
};
if db_lock.applications.game_statuses.contains_key(&game_id) {
if db_lock.applications.game_statuses.contains_key(&drop_data.game_id) {
continue;
}