LeonMMcoset/leonwww
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

i didnt test this a lot but i added progress bars to GURT:// based downloads 🎉

Browse Source
This commit is contained in:
Gabriella Casap
2025-09-26 14:49:22 +01:00
parent 5e4c381b84
commit 04434f176b
6 changed files with 448 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
flumi/Scripts/Browser/DownloadManager.gd
View File

@@ -85,7 +85,7 @@ func _start_download(download_id: String, url: String, save_path: String, downlo
} }
if url.begins_with("gurt://"): if url.begins_with("gurt://"):
_download_gurt_resource(download_id, url) _start_gurt_download(download_id, url)
else: else:
_start_http_download(download_id, url) _start_http_download(download_id, url)
@@ -122,56 +122,115 @@ func _start_http_download(download_id: String, url: String):
var timer = Timer.new() var timer = Timer.new()
timer.name = "ProgressTimer_" + download_id timer.name = "ProgressTimer_" + download_id
timer.wait_time = 0.5 timer.wait_time = 0.2
timer.timeout.connect(func(): _update_download_progress(download_id)) timer.timeout.connect(func(): _update_download_progress(download_id))
main_node.add_child(timer) main_node.add_child(timer)
timer.start() timer.start()
func _download_gurt_resource(download_id: String, url: String): func _start_gurt_download(download_id: String, url: String):
if not active_downloads.has(download_id): if not active_downloads.has(download_id):
return return
var progress_ui = active_downloads[download_id]["progress_ui"] var progress_ui = active_downloads[download_id]["progress_ui"]
var save_path = active_downloads[download_id]["save_path"] var save_path = active_downloads[download_id]["save_path"]
if progress_ui: var client = GurtProtocolClient.new()
progress_ui.update_progress(0, 0, -1) # -1 indicates unknown total size for ca in CertificateManager.trusted_ca_certificates:
client.add_ca_certificate(ca)
if not client.create_client_with_dns(30, GurtProtocol.DNS_SERVER_IP, GurtProtocol.DNS_SERVER_PORT):
if progress_ui:
progress_ui.set_error("Failed to create GURT client")
active_downloads.erase(download_id)
return
var resource_data = await Network.fetch_gurt_resource(url, true) active_downloads[download_id]["gurt_client"] = client
var started_cb = Callable(self, "_on_gurt_download_started")
if not client.download_started.is_connected(started_cb):
client.download_started.connect(started_cb)
var progress_cb = Callable(self, "_on_gurt_download_progress")
if not client.download_progress.is_connected(progress_cb):
client.download_progress.connect(progress_cb)
var completed_cb = Callable(self, "_on_gurt_download_completed")
if not client.download_completed.is_connected(completed_cb):
client.download_completed.connect(completed_cb)
var failed_cb = Callable(self, "_on_gurt_download_failed")
if not client.download_failed.is_connected(failed_cb):
client.download_failed.connect(failed_cb)
client.start_download(download_id, url, save_path)
var poll_timer = Timer.new()
poll_timer.wait_time = 0.2
poll_timer.one_shot = false
poll_timer.name = "GurtPoll_" + download_id
poll_timer.timeout.connect(func():
if not active_downloads.has(download_id):
poll_timer.queue_free()
return
var c = active_downloads[download_id].get("gurt_client", null)
if c:
c.poll_events()
else:
poll_timer.queue_free()
)
main_node.add_child(poll_timer)
poll_timer.start()
func _on_gurt_download_started(download_id: String, total_bytes: int):
if not active_downloads.has(download_id): if not active_downloads.has(download_id):
return return
var info = active_downloads[download_id]
info.total_bytes = max(total_bytes, 0)
info.downloaded_bytes = 0
var ui = info.progress_ui
if ui:
ui.update_progress(0.0, 0, info.total_bytes)
if resource_data.is_empty(): func _on_gurt_download_progress(download_id: String, downloaded_bytes: int, total_bytes: int):
var error_msg = "Failed to fetch gurt:// resource" if not active_downloads.has(download_id):
print(error_msg)
if progress_ui:
progress_ui.set_error(error_msg)
active_downloads.erase(download_id)
return return
var info = active_downloads[download_id]
if total_bytes > 0:
info.total_bytes = total_bytes
info.downloaded_bytes = downloaded_bytes
var total = info.total_bytes
var p = 0.0
if total > 0:
p = float(downloaded_bytes) / float(total) * 100.0
var ui = info.progress_ui
if ui:
ui.update_progress(p, downloaded_bytes, total)
var file = FileAccess.open(save_path, FileAccess.WRITE) func _on_gurt_download_completed(download_id: String, save_path: String):
if not file: if not active_downloads.has(download_id):
var error_msg = "Failed to create download file: " + save_path
print(error_msg)
if progress_ui:
progress_ui.set_error(error_msg)
active_downloads.erase(download_id)
return return
var info = active_downloads[download_id]
var path = save_path if not save_path.is_empty() else info.save_path
var size = 0
if FileAccess.file_exists(path):
var f = FileAccess.open(path, FileAccess.READ)
if f:
size = f.get_length()
f.close()
info.total_bytes = size
info.downloaded_bytes = size
var ui = info.progress_ui
if ui:
ui.set_completed(path)
_add_to_download_history(info, size, path)
active_downloads.erase(download_id)
file.store_buffer(resource_data) func _on_gurt_download_failed(download_id: String, message: String):
file.close() if not active_downloads.has(download_id):
return
var file_size = resource_data.size() var info = active_downloads[download_id]
var ui = info.progress_ui
active_downloads[download_id]["total_bytes"] = file_size if ui:
active_downloads[download_id]["downloaded_bytes"] = file_size ui.set_error(message)
var path = info.save_path
if progress_ui: if FileAccess.file_exists(path):
progress_ui.set_completed(save_path) DirAccess.remove_absolute(path)
_add_to_download_history(active_downloads[download_id], file_size, save_path)
active_downloads.erase(download_id) active_downloads.erase(download_id)
func _update_download_progress(download_id: String): func _update_download_progress(download_id: String):
@@ -240,6 +299,10 @@ func _on_download_progress_cancelled(download_id: String):
return return
var download_info = active_downloads[download_id] var download_info = active_downloads[download_id]
if download_info.has("gurt_client"):
var c = download_info["gurt_client"]
c.cancel_download(download_id)
return
var http_request = download_info.get("http_request", null) var http_request = download_info.get("http_request", null)
if http_request: if http_request:

4
flumi/Scripts/Browser/NetworkRequest.gd
View File

@@ -211,8 +211,10 @@ static func format_bytes(given_size: int) -> String:
return str(given_size) + " B" return str(given_size) + " B"
elif given_size < 1024 * 1024: elif given_size < 1024 * 1024:
return str(given_size / 1024) + " KB" return str(given_size / 1024) + " KB"
elif given_size < 1024 * 1024 * 1024:
return "%.1f MB" % (given_size / (1024.0 * 1024.0))
else: else:
return str(given_size / (1024.0 * 1024)) + " MB" return "%.2f GB" % (given_size / (1024.0 * 1024.0 * 1024.0))
func get_time_display() -> String: func get_time_display() -> String:
if status == RequestStatus.PENDING: if status == RequestStatus.PENDING:

258
protocol/gdextension/src/lib.rs
View File

@@ -1,9 +1,13 @@
use godot::prelude::*; use godot::prelude::*;
use gurtlib::prelude::*; use gurtlib::prelude::*;
use gurtlib::{GurtMethod, GurtClientConfig, GurtRequest}; use gurtlib::{GurtMethod, GurtClientConfig, GurtRequest, GurtResponseHead};
use tokio::runtime::Runtime;
use std::sync::Arc;
use std::cell::RefCell; use std::cell::RefCell;
use std::fs::File;
use std::io::Write;
use std::sync::Arc;
use std::sync::Mutex;
use std::collections::HashMap;
use tokio::runtime::Runtime;
struct GurtGodotExtension; struct GurtGodotExtension;
@@ -14,32 +18,34 @@ unsafe impl ExtensionLibrary for GurtGodotExtension {}
#[class(init)] #[class(init)]
struct GurtProtocolClient { struct GurtProtocolClient {
base: Base<RefCounted>, base: Base<RefCounted>,
client: Arc<RefCell<Option<GurtClient>>>, client: Arc<RefCell<Option<GurtClient>>>,
runtime: Arc<RefCell<Option<Runtime>>>, runtime: Arc<RefCell<Option<Runtime>>>,
ca_certificates: Arc<RefCell<Vec<String>>>, ca_certificates: Arc<RefCell<Vec<String>>>,
cancel_flags: Arc<Mutex<HashMap<String, bool>>>,
event_queue: Arc<Mutex<Vec<DownloadEvent>>>,
} }
#[derive(GodotClass)] #[derive(GodotClass)]
#[class(init)] #[class(init)]
struct GurtGDResponse { struct GurtGDResponse {
base: Base<RefCounted>, base: Base<RefCounted>,
#[var] #[var]
status_code: i32, status_code: i32,
#[var] #[var]
status_message: GString, status_message: GString,
#[var] #[var]
headers: Dictionary, headers: Dictionary,
#[var] #[var]
is_success: bool, is_success: bool,
#[var] #[var]
body: PackedByteArray, // Raw bytes body: PackedByteArray, // Raw bytes
#[var] #[var]
text: GString, // Decoded text text: GString, // Decoded text
} }
@@ -59,7 +65,7 @@ impl GurtGDResponse {
content_type.starts_with("video/") || content_type.starts_with("video/") ||
content_type.starts_with("audio/") content_type.starts_with("audio/")
} }
#[func] #[func]
fn is_text(&self) -> bool { fn is_text(&self) -> bool {
let content_type = self.get_header("content-type".into()).to_string(); let content_type = self.get_header("content-type".into()).to_string();
@@ -68,7 +74,7 @@ impl GurtGDResponse {
content_type.starts_with("application/xml") || content_type.starts_with("application/xml") ||
content_type.is_empty() content_type.is_empty()
} }
#[func] #[func]
fn debug_info(&self) -> GString { fn debug_info(&self) -> GString {
let content_length = self.get_header("content-length".into()).to_string(); let content_length = self.get_header("content-length".into()).to_string();
@@ -93,6 +99,17 @@ struct GurtProtocolServer {
base: Base<RefCounted>, base: Base<RefCounted>,
} }
struct DLState { file: Option<std::fs::File>, total_bytes: i64, downloaded: i64 }
#[derive(Clone)]
enum DownloadEvent {
Started(String, i64),
Progress(String, i64, i64),
Completed(String, String),
Failed(String, String),
}
#[godot_api] #[godot_api]
impl GurtProtocolClient { impl GurtProtocolClient {
fn init(base: Base<RefCounted>) -> Self { fn init(base: Base<RefCounted>) -> Self {
@@ -101,13 +118,27 @@ impl GurtProtocolClient {
client: Arc::new(RefCell::new(None)), client: Arc::new(RefCell::new(None)),
runtime: Arc::new(RefCell::new(None)), runtime: Arc::new(RefCell::new(None)),
ca_certificates: Arc::new(RefCell::new(Vec::new())), ca_certificates: Arc::new(RefCell::new(Vec::new())),
cancel_flags: Arc::new(Mutex::new(HashMap::new())),
event_queue: Arc::new(Mutex::new(Vec::new())),
} }
} }
#[signal] #[signal]
fn request_completed(response: Gd<GurtGDResponse>); fn request_completed(response: Gd<GurtGDResponse>);
#[func] #[signal]
fn download_started(download_id: GString, total_bytes: i64);
#[signal]
fn download_progress(download_id: GString, downloaded_bytes: i64, total_bytes: i64);
#[signal]
fn download_completed(download_id: GString, save_path: GString);
#[signal]
fn download_failed(download_id: GString, message: GString);
#[func]
fn create_client(&mut self, timeout_seconds: i32) -> bool { fn create_client(&mut self, timeout_seconds: i32) -> bool {
let runtime = match Runtime::new() { let runtime = match Runtime::new() {
Ok(rt) => rt, Ok(rt) => rt,
@@ -116,21 +147,21 @@ impl GurtProtocolClient {
return false; return false;
} }
}; };
let mut config = GurtClientConfig::default(); let mut config = GurtClientConfig::default();
config.request_timeout = tokio::time::Duration::from_secs(timeout_seconds as u64); config.request_timeout = tokio::time::Duration::from_secs(timeout_seconds as u64);
// Add custom CA certificates // Add custom CA certificates
config.custom_ca_certificates = self.ca_certificates.borrow().clone(); config.custom_ca_certificates = self.ca_certificates.borrow().clone();
let client = GurtClient::with_config(config); let client = GurtClient::with_config(config);
*self.runtime.borrow_mut() = Some(runtime); *self.runtime.borrow_mut() = Some(runtime);
*self.client.borrow_mut() = Some(client); *self.client.borrow_mut() = Some(client);
true true
} }
#[func] #[func]
fn create_client_with_dns(&mut self, timeout_seconds: i32, dns_ip: GString, dns_port: i32) -> bool { fn create_client_with_dns(&mut self, timeout_seconds: i32, dns_ip: GString, dns_port: i32) -> bool {
let runtime = match Runtime::new() { let runtime = match Runtime::new() {
@@ -140,22 +171,22 @@ impl GurtProtocolClient {
return false; return false;
} }
}; };
let mut config = GurtClientConfig::default(); let mut config = GurtClientConfig::default();
config.request_timeout = tokio::time::Duration::from_secs(timeout_seconds as u64); config.request_timeout = tokio::time::Duration::from_secs(timeout_seconds as u64);
config.dns_server_ip = dns_ip.to_string(); config.dns_server_ip = dns_ip.to_string();
config.dns_server_port = dns_port as u16; config.dns_server_port = dns_port as u16;
config.custom_ca_certificates = self.ca_certificates.borrow().clone(); config.custom_ca_certificates = self.ca_certificates.borrow().clone();
let client = GurtClient::with_config(config); let client = GurtClient::with_config(config);
*self.runtime.borrow_mut() = Some(runtime); *self.runtime.borrow_mut() = Some(runtime);
*self.client.borrow_mut() = Some(client); *self.client.borrow_mut() = Some(client);
true true
} }
#[func] #[func]
fn request(&self, url: GString, options: Dictionary) -> Option<Gd<GurtGDResponse>> { fn request(&self, url: GString, options: Dictionary) -> Option<Gd<GurtGDResponse>> {
let runtime_binding = self.runtime.borrow(); let runtime_binding = self.runtime.borrow();
@@ -166,9 +197,9 @@ impl GurtProtocolClient {
return None; return None;
} }
}; };
let url_str = url.to_string(); let url_str = url.to_string();
// Parse URL to get host and port // Parse URL to get host and port
let parsed_url = match url::Url::parse(&url_str) { let parsed_url = match url::Url::parse(&url_str) {
Ok(u) => u, Ok(u) => u,
@@ -187,17 +218,17 @@ impl GurtProtocolClient {
}; };
let port = parsed_url.port().unwrap_or(4878); let port = parsed_url.port().unwrap_or(4878);
let path_with_query = if parsed_url.path().is_empty() { let path_with_query = if parsed_url.path().is_empty() {
"/" "/"
} else { } else {
parsed_url.path() parsed_url.path()
}; };
let path = match parsed_url.query() { let path = match parsed_url.query() {
Some(query) => format!("{}?{}", path_with_query, query), Some(query) => format!("{}?{}", path_with_query, query),
None => path_with_query.to_string(), None => path_with_query.to_string(),
}; };
let method_str = options.get("method").unwrap_or("GET".to_variant()).to::<String>(); let method_str = options.get("method").unwrap_or("GET".to_variant()).to::<String>();
let method = match method_str.to_uppercase().as_str() { let method = match method_str.to_uppercase().as_str() {
"GET" => GurtMethod::GET, "GET" => GurtMethod::GET,
@@ -212,7 +243,7 @@ impl GurtProtocolClient {
GurtMethod::GET GurtMethod::GET
} }
}; };
let client_binding = self.client.borrow(); let client_binding = self.client.borrow();
let client = match client_binding.as_ref() { let client = match client_binding.as_ref() {
Some(c) => c, Some(c) => c,
@@ -221,13 +252,13 @@ impl GurtProtocolClient {
return None; return None;
} }
}; };
let body = options.get("body").unwrap_or("".to_variant()).to::<String>(); let body = options.get("body").unwrap_or("".to_variant()).to::<String>();
let headers_dict = options.get("headers").unwrap_or(Dictionary::new().to_variant()).to::<Dictionary>(); let headers_dict = options.get("headers").unwrap_or(Dictionary::new().to_variant()).to::<Dictionary>();
let mut request = GurtRequest::new(method, path.to_string()) let mut request = GurtRequest::new(method, path.to_string())
.with_header("User-Agent", "GURT-Client/1.0.0"); .with_header("User-Agent", "GURT-Client/1.0.0");
for key_variant in headers_dict.keys_array().iter_shared() { for key_variant in headers_dict.keys_array().iter_shared() {
let key = key_variant.to::<String>(); let key = key_variant.to::<String>();
if let Some(value_variant) = headers_dict.get(key_variant) { if let Some(value_variant) = headers_dict.get(key_variant) {
@@ -235,11 +266,11 @@ impl GurtProtocolClient {
request = request.with_header(key, value); request = request.with_header(key, value);
} }
} }
if !body.is_empty() { if !body.is_empty() {
request = request.with_string_body(&body); request = request.with_string_body(&body);
} }
let response = match runtime.block_on(async { let response = match runtime.block_on(async {
client.send_request(host, port, request).await client.send_request(host, port, request).await
}) { }) {
@@ -249,66 +280,189 @@ impl GurtProtocolClient {
return None; return None;
} }
}; };
Some(self.convert_response(response)) Some(self.convert_response(response))
} }
#[func]
fn start_download(&mut self, download_id: GString, url: GString, save_path: GString) -> bool {
let runtime_handle = {
let runtime_binding = self.runtime.borrow();
match runtime_binding.as_ref() {
Some(rt) => rt.handle().clone(),
None => { godot_print!("No runtime available"); return false; }
}
};
let client_instance = {
let client_binding = self.client.borrow();
match client_binding.as_ref() {
Some(c) => c.clone(),
None => { godot_print!("No client available"); return false; }
}
};
let url_str = url.to_string();
let save_path_str = save_path.to_string();
let download_id_string = download_id.to_string();
let cancel_flags = self.cancel_flags.clone();
let event_queue = self.event_queue.clone();
runtime_handle.spawn(async move {
let event_queue_main = event_queue.clone();
let parsed_url = match url::Url::parse(&url_str) { Ok(u) => u, Err(e) => {
if let Ok(mut q) = event_queue.lock() { q.push(DownloadEvent::Failed(download_id_string.clone(), format!("Invalid URL: {}", e))); }
return;
}};
let host = match parsed_url.host_str() { Some(h) => h.to_string(), None => {
if let Ok(mut q) = event_queue.lock() { q.push(DownloadEvent::Failed(download_id_string.clone(), "URL must have a host".to_string())); }
return;
}};
let port = parsed_url.port().unwrap_or(4878);
let path_with_query = if parsed_url.path().is_empty() { "/".to_string() } else { parsed_url.path().to_string() };
let path = match parsed_url.query() { Some(query) => format!("{}?{}", path_with_query, query), None => path_with_query };
let state = Arc::new(Mutex::new(DLState { file: None, total_bytes: -1, downloaded: 0 }));
let request = GurtRequest::new(GurtMethod::GET, path).with_header("User-Agent", "GURT-Client/1.0.0");
let state_head = state.clone();
let event_queue_head = event_queue.clone();
let id_for_head = download_id_string.clone();
let sp_for_head = save_path_str.clone();
let on_head = move |head: &GurtResponseHead| {
if head.status_code < 200 || head.status_code >= 300 {
if let Ok(mut q) = event_queue_head.lock() { q.push(DownloadEvent::Failed(id_for_head.clone(), format!("{} {}", head.status_code, head.status_message))); }
return;
}
let mut total: i64 = -1;
if let Some(cl) = head.headers.get("content-length").or_else(|| head.headers.get("Content-Length")) {
if let Ok(v) = cl.parse::<i64>() { total = v; }
}
match File::create(&sp_for_head) {
Ok(f) => {
if let Ok(mut st) = state_head.lock() { st.file = Some(f); st.total_bytes = total; }
}
Err(e) => { if let Ok(mut q) = event_queue_head.lock() { q.push(DownloadEvent::Failed(id_for_head.clone(), format!("File error: {}", e))); } }
}
if let Ok(mut q) = event_queue_head.lock() { q.push(DownloadEvent::Started(id_for_head.clone(), total)); }
};
let state_chunk = state.clone();
let event_queue_chunk = event_queue.clone();
let id_for_chunk = download_id_string.clone();
let on_chunk = move |chunk: &[u8]| -> bool {
if let Ok(map) = cancel_flags.lock() {
if map.get(&id_for_chunk).copied().unwrap_or(false) { return false; }
}
let mut down = 0i64; let mut total = -1i64; let mut write_result: std::io::Result<()> = Ok(());
if let Ok(mut st) = state_chunk.lock() {
if let Some(f) = st.file.as_mut() { write_result = f.write_all(chunk); }
st.downloaded += chunk.len() as i64; down = st.downloaded; total = st.total_bytes;
}
if let Err(e) = write_result { if let Ok(mut q) = event_queue_chunk.lock() { q.push(DownloadEvent::Failed(id_for_chunk.clone(), format!("Write error: {}", e))); } return false; }
if let Ok(mut q) = event_queue_chunk.lock() { q.push(DownloadEvent::Progress(id_for_chunk.clone(), down, total)); }
true
};
let result = client_instance.stream_request(host.as_str(), port, request, on_head, on_chunk).await;
match result {
Ok(()) => {
if let Ok(mut st) = state.lock() { if let Some(f) = st.file.as_mut() { let _ = f.flush(); } }
if let Ok(mut q) = event_queue_main.lock() { q.push(DownloadEvent::Completed(download_id_string.clone(), save_path_str.clone())); }
}
Err(e) => {
if let Ok(mut q) = event_queue_main.lock() { q.push(DownloadEvent::Failed(download_id_string.clone(), format!("{}", e))); }
}
}
});
true
}
#[func]
fn cancel_download(&mut self, download_id: GString) {
if let Ok(mut map) = self.cancel_flags.lock() { map.insert(download_id.to_string(), true); }
}
#[func]
fn poll_events(&mut self) {
let mut drained: Vec<DownloadEvent> = Vec::new();
if let Ok(mut q) = self.event_queue.lock() { drained.append(&mut *q); }
for ev in drained.into_iter() {
match ev {
DownloadEvent::Started(id, total) => { let mut owner = self.base.to_gd(); let args = [GString::from(id).to_variant(), (total as i64).to_variant()]; owner.emit_signal("download_started".into(), &args); }
DownloadEvent::Progress(id, down, total) => { let mut owner = self.base.to_gd(); let args = [GString::from(id).to_variant(), (down as i64).to_variant(), (total as i64).to_variant()]; owner.emit_signal("download_progress".into(), &args); }
DownloadEvent::Completed(id, path) => { let mut owner = self.base.to_gd(); let args = [GString::from(id).to_variant(), GString::from(path).to_variant()]; owner.emit_signal("download_completed".into(), &args); }
DownloadEvent::Failed(id, msg) => { let mut owner = self.base.to_gd(); let args = [GString::from(id).to_variant(), GString::from(msg).to_variant()]; owner.emit_signal("download_failed".into(), &args); }
}
}
}
#[func] #[func]
fn disconnect(&mut self) { fn disconnect(&mut self) {
*self.client.borrow_mut() = None; *self.client.borrow_mut() = None;
*self.runtime.borrow_mut() = None; *self.runtime.borrow_mut() = None;
} }
#[func] #[func]
fn is_connected(&self) -> bool { fn is_connected(&self) -> bool {
self.client.borrow().is_some() self.client.borrow().is_some()
} }
#[func] #[func]
fn get_version(&self) -> GString { fn get_version(&self) -> GString {
gurtlib::GURT_VERSION.to_string().into() gurtlib::GURT_VERSION.to_string().into()
} }
#[func] #[func]
fn get_default_port(&self) -> i32 { fn get_default_port(&self) -> i32 {
gurtlib::DEFAULT_PORT as i32 gurtlib::DEFAULT_PORT as i32
} }
#[func] #[func]
fn add_ca_certificate(&self, cert_pem: GString) { fn add_ca_certificate(&self, cert_pem: GString) {
self.ca_certificates.borrow_mut().push(cert_pem.to_string()); self.ca_certificates.borrow_mut().push(cert_pem.to_string());
} }
#[func] #[func]
fn clear_ca_certificates(&self) { fn clear_ca_certificates(&self) {
self.ca_certificates.borrow_mut().clear(); self.ca_certificates.borrow_mut().clear();
} }
#[func] #[func]
fn get_ca_certificate_count(&self) -> i32 { fn get_ca_certificate_count(&self) -> i32 {
self.ca_certificates.borrow().len() as i32 self.ca_certificates.borrow().len() as i32
} }
fn emit_download_failed(&mut self, download_id: &GString, message: String) {
let mut owner = self.base.to_gd();
let args = [
download_id.to_variant(),
GString::from(message).to_variant(),
];
owner.emit_signal("download_failed".into(), &args);
}
fn convert_response(&self, response: GurtResponse) -> Gd<GurtGDResponse> { fn convert_response(&self, response: GurtResponse) -> Gd<GurtGDResponse> {
let mut gd_response = GurtGDResponse::new_gd(); let mut gd_response = GurtGDResponse::new_gd();
gd_response.bind_mut().status_code = response.status_code as i32; gd_response.bind_mut().status_code = response.status_code as i32;
gd_response.bind_mut().status_message = response.status_message.clone().into(); gd_response.bind_mut().status_message = response.status_message.clone().into();
gd_response.bind_mut().is_success = response.is_success(); gd_response.bind_mut().is_success = response.is_success();
let mut headers = Dictionary::new(); let mut headers = Dictionary::new();
for (key, value) in &response.headers { for (key, value) in &response.headers {
headers.set(key.clone(), value.clone()); headers.set(key.clone(), value.clone());
} }
gd_response.bind_mut().headers = headers; gd_response.bind_mut().headers = headers;
let mut body = PackedByteArray::new(); let mut body = PackedByteArray::new();
body.resize(response.body.len()); body.resize(response.body.len());
for (i, byte) in response.body.iter().enumerate() { for (i, byte) in response.body.iter().enumerate() {
body[i] = *byte; body[i] = *byte;
} }
gd_response.bind_mut().body = body; gd_response.bind_mut().body = body;
match std::str::from_utf8(&response.body) { match std::str::from_utf8(&response.body) {
Ok(text_str) => { Ok(text_str) => {
gd_response.bind_mut().text = text_str.into(); gd_response.bind_mut().text = text_str.into();
@@ -319,7 +473,7 @@ impl GurtProtocolClient {
gd_response.bind_mut().text = format!("[Binary data: {} ({} bytes)]", content_type, size).into(); gd_response.bind_mut().text = format!("[Binary data: {} ({} bytes)]", content_type, size).into();
} }
} }
gd_response gd_response
} }
} }

135
protocol/library/src/client.rs
View File

@@ -527,6 +527,139 @@ impl GurtClient {
Ok((host, port, path)) Ok((host, port, path))
} }
pub async fn stream_request<HeadCb, ChunkCb>(&self,
host: &str,
port: u16,
mut request: GurtRequest,
mut on_head: HeadCb,
mut on_chunk: ChunkCb,
) -> Result<()>
where
HeadCb: FnMut(&crate::message::GurtResponseHead) + Send,
ChunkCb: FnMut(&[u8]) -> bool + Send,
{
let resolved_host = self.resolve_domain(host).await?;
request = request.with_header("Host", host);
let mut tls_stream = self.get_pooled_connection(&resolved_host, port, Some(host)).await?;
let request_data = request.to_string();
tls_stream.write_all(request_data.as_bytes()).await
.map_err(|e| GurtError::connection(format!("Failed to write request: {}", e)))?;
let mut buffer: Vec<u8> = Vec::new();
let mut temp_buffer = [0u8; 8192];
let start_time = std::time::Instant::now();
let mut headers_parsed = false;
let mut expected_body_length: Option<usize> = None;
let mut headers_end_pos: Option<usize> = None;
let mut head_emitted = false;
let mut delivered: usize = 0;
loop {
if start_time.elapsed() > self.config.request_timeout {
return Err(GurtError::timeout("Request timeout"));
}
match timeout(Duration::from_millis(400), tls_stream.read(&mut temp_buffer)).await {
Ok(Ok(0)) => {
if headers_parsed && !head_emitted {
let head = crate::message::GurtResponseHead {
version: String::new(),
status_code: 0,
status_message: String::new(),
headers: std::collections::HashMap::new(),
};
on_head(&head);
head_emitted = true;
}
break;
}
Ok(Ok(n)) => {
buffer.extend_from_slice(&temp_buffer[..n]);
if !headers_parsed {
if let Some(pos) = buffer.windows(4).position(|w| w == b"\r\n\r\n") {
headers_end_pos = Some(pos + 4);
headers_parsed = true;
let headers_section = std::str::from_utf8(&buffer[..pos])
.map_err(|e| GurtError::invalid_message(format!("Invalid UTF-8 in headers: {}", e)))?;
let mut lines = headers_section.split("\r\n");
let status_line = lines.next().unwrap_or("");
let parts: Vec<&str> = status_line.splitn(3, ' ').collect();
let mut version = String::new();
let mut status_code: u16 = 0;
let mut status_message = String::new();
if parts.len() >= 2 {
version = parts[0].to_string();
status_code = parts[1].parse().unwrap_or(0);
if parts.len() > 2 { status_message = parts[2].to_string(); }
}
let mut headers = std::collections::HashMap::new();
for line in lines {
if line.is_empty() { break; }
if let Some(colon) = line.find(':') {
let key = line[..colon].trim().to_lowercase();
let value = line[colon+1..].trim().to_string();
if key == "content-length" { expected_body_length = value.parse().ok(); }
headers.insert(key, value);
}
}
let head = crate::message::GurtResponseHead {
version,
status_code,
status_message,
headers,
};
on_head(&head);
head_emitted = true;
if let Some(end) = headers_end_pos {
if buffer.len() > end {
let body_slice = &buffer[end..];
if !on_chunk(body_slice) {
return Err(GurtError::Cancelled);
}
delivered = body_slice.len();
}
}
}
} else {
if let Some(end) = headers_end_pos {
let available = buffer.len().saturating_sub(end + delivered);
if available > 0 {
let start = end + delivered;
let end_pos = end + delivered + available;
if !on_chunk(&buffer[start..end_pos]) {
return Err(GurtError::Cancelled);
}
delivered += available;
}
if let Some(expected_len) = expected_body_length {
if delivered >= expected_len { break; }
}
}
}
}
Ok(Err(e)) => return Err(GurtError::connection(format!("Read error: {}", e))),
Err(_) => continue,
}
}
if let (Some(end), Some(expected_len)) = (headers_end_pos, expected_body_length) {
if delivered >= expected_len {
self.return_connection_to_pool(&resolved_host, port, tls_stream);
}
}
Ok(())
}
async fn resolve_domain(&self, domain: &str) -> Result<String> { async fn resolve_domain(&self, domain: &str) -> Result<String> {
match self.dns_cache.lock() { match self.dns_cache.lock() {
@@ -732,4 +865,4 @@ mod tests {
assert!(handshake_request.headers.contains_key("user-agent")); assert!(handshake_request.headers.contains_key("user-agent"));
} }
} }

4
protocol/library/src/lib.rs
View File

@@ -6,7 +6,7 @@ pub mod error;
pub mod message; pub mod message;
pub use error::{GurtError, Result}; pub use error::{GurtError, Result};
pub use message::{GurtMessage, GurtRequest, GurtResponse, GurtMethod}; pub use message::{GurtMessage, GurtRequest, GurtResponse, GurtResponseHead, GurtMethod};
pub use protocol::{GurtStatusCode, GURT_VERSION, DEFAULT_PORT}; pub use protocol::{GurtStatusCode, GURT_VERSION, DEFAULT_PORT};
pub use crypto::{CryptoManager, TlsConfig, GURT_ALPN, TLS_VERSION}; pub use crypto::{CryptoManager, TlsConfig, GURT_ALPN, TLS_VERSION};
pub use server::{GurtServer, GurtHandler, ServerContext, Route}; pub use server::{GurtServer, GurtHandler, ServerContext, Route};
@@ -15,7 +15,7 @@ pub use client::{GurtClient, GurtClientConfig};
pub mod prelude { pub mod prelude {
pub use crate::{ pub use crate::{
GurtError, Result, GurtError, Result,
GurtMessage, GurtRequest, GurtResponse, GurtMessage, GurtRequest, GurtResponse, GurtResponseHead,
GURT_VERSION, DEFAULT_PORT, GURT_VERSION, DEFAULT_PORT,
CryptoManager, TlsConfig, GURT_ALPN, TLS_VERSION, CryptoManager, TlsConfig, GURT_ALPN, TLS_VERSION,
GurtServer, GurtHandler, ServerContext, Route, GurtServer, GurtHandler, ServerContext, Route,

8
protocol/library/src/message.rs
View File

@@ -225,6 +225,14 @@ pub struct GurtResponse {
pub body: Vec<u8>, pub body: Vec<u8>,
} }
#[derive(Debug, Clone)]
pub struct GurtResponseHead {
pub version: String,
pub status_code: u16,
pub status_message: String,
pub headers: GurtHeaders,
}
impl GurtResponse { impl GurtResponse {
pub fn new(status_code: GurtStatusCode) -> Self { pub fn new(status_code: GurtStatusCode) -> Self {
Self { Self {