Merge pull request #63740 from Faless/net/4.x_http_request_decompress

[HTTP] Implement streaming decompression.

2 files changed, 67 insertions, 62 deletions

diff --git a/scene/main/http_request.cpp b/scene/main/http_request.cpp
index bec378dd91..2c395ec07d 100644
--- a/scene/main/http_request.cpp
+++ b/scene/main/http_request.cpp
@@ -32,9 +32,6 @@
 #include "core/io/compression.h"
 #include "scene/main/timer.h"
 
-void HTTPRequest::_redirect_request(const String &p_new_url) {
-}
-
 Error HTTPRequest::_request() {
 	return client->connect_to_host(url, port, use_tls, validate_tls);
 }
@@ -48,6 +45,7 @@ Error HTTPRequest::_parse_url(const String &p_url) {
 	body_len = -1;
 	body.clear();
 	downloaded.set(0);
+	final_body_size.set(0);
 	redirections = 0;
 
 	String scheme;
@@ -153,7 +151,7 @@ Error HTTPRequest::request_raw(const String &p_url, const Vector<String> &p_cust
 		client->set_blocking_mode(false);
 		err = _request();
 		if (err != OK) {
-			call_deferred(SNAME("_request_done"), RESULT_CANT_CONNECT, 0, PackedStringArray(), PackedByteArray());
+			_defer_done(RESULT_CANT_CONNECT, 0, PackedStringArray(), PackedByteArray());
 			return ERR_CANT_CONNECT;
 		}
 
@@ -169,7 +167,7 @@ void HTTPRequest::_thread_func(void *p_userdata) {
 	Error err = hr->_request();
 
 	if (err != OK) {
-		hr->call_deferred(SNAME("_request_done"), RESULT_CANT_CONNECT, 0, PackedStringArray(), PackedByteArray());
+		hr->_defer_done(RESULT_CANT_CONNECT, 0, PackedStringArray(), PackedByteArray());
 	} else {
 		while (!hr->thread_request_quit.is_set()) {
 			bool exit = hr->_update_connection();
@@ -198,6 +196,7 @@ void HTTPRequest::cancel_request() {
 	}
 
 	file.unref();
+	decompressor.unref();
 	client->close();
 	body.clear();
 	got_response = false;
@@ -208,7 +207,7 @@ void HTTPRequest::cancel_request() {
 
 bool HTTPRequest::_handle_response(bool *ret_value) {
 	if (!client->has_response()) {
-		call_deferred(SNAME("_request_done"), RESULT_NO_RESPONSE, 0, PackedStringArray(), PackedByteArray());
+		_defer_done(RESULT_NO_RESPONSE, 0, PackedStringArray(), PackedByteArray());
 		*ret_value = true;
 		return true;
 	}
@@ -219,6 +218,9 @@ bool HTTPRequest::_handle_response(bool *ret_value) {
 	client->get_response_headers(&rheaders);
 	response_headers.clear();
 	downloaded.set(0);
+	final_body_size.set(0);
+	decompressor.unref();
+
 	for (const String &E : rheaders) {
 		response_headers.push_back(E);
 	}
@@ -227,7 +229,7 @@ bool HTTPRequest::_handle_response(bool *ret_value) {
 		// Handle redirect.
 
 		if (max_redirects >= 0 && redirections >= max_redirects) {
-			call_deferred(SNAME("_request_done"), RESULT_REDIRECT_LIMIT_REACHED, response_code, response_headers, PackedByteArray());
+			_defer_done(RESULT_REDIRECT_LIMIT_REACHED, response_code, response_headers, PackedByteArray());
 			*ret_value = true;
 			return true;
 		}
@@ -259,6 +261,7 @@ bool HTTPRequest::_handle_response(bool *ret_value) {
 				body_len = -1;
 				body.clear();
 				downloaded.set(0);
+				final_body_size.set(0);
 				redirections = new_redirs;
 				*ret_value = false;
 				return true;
@@ -266,13 +269,26 @@ bool HTTPRequest::_handle_response(bool *ret_value) {
 		}
 	}
 
+	// Check if we need to start streaming decompression.
+	String content_encoding;
+	if (accept_gzip) {
+		content_encoding = get_header_value(response_headers, "Content-Encoding").to_lower();
+	}
+	if (content_encoding == "gzip") {
+		decompressor.instantiate();
+		decompressor->start_decompression(false, get_download_chunk_size() * 2);
+	} else if (content_encoding == "deflate") {
+		decompressor.instantiate();
+		decompressor->start_decompression(true, get_download_chunk_size() * 2);
+	}
+
 	return false;
 }
 
 bool HTTPRequest::_update_connection() {
 	switch (client->get_status()) {
 		case HTTPClient::STATUS_DISCONNECTED: {
-			call_deferred(SNAME("_request_done"), RESULT_CANT_CONNECT, 0, PackedStringArray(), PackedByteArray());
+			_defer_done(RESULT_CANT_CONNECT, 0, PackedStringArray(), PackedByteArray());
 			return true; // End it, since it's disconnected.
 		} break;
 		case HTTPClient::STATUS_RESOLVING: {
@@ -281,7 +297,7 @@ bool HTTPRequest::_update_connection() {
 			return false;
 		} break;
 		case HTTPClient::STATUS_CANT_RESOLVE: {
-			call_deferred(SNAME("_request_done"), RESULT_CANT_RESOLVE, 0, PackedStringArray(), PackedByteArray());
+			_defer_done(RESULT_CANT_RESOLVE, 0, PackedStringArray(), PackedByteArray());
 			return true;
 
 		} break;
@@ -291,7 +307,7 @@ bool HTTPRequest::_update_connection() {
 			return false;
 		} break; // Connecting to IP.
 		case HTTPClient::STATUS_CANT_CONNECT: {
-			call_deferred(SNAME("_request_done"), RESULT_CANT_CONNECT, 0, PackedStringArray(), PackedByteArray());
+			_defer_done(RESULT_CANT_CONNECT, 0, PackedStringArray(), PackedByteArray());
 			return true;
 
 		} break;
@@ -306,16 +322,16 @@ bool HTTPRequest::_update_connection() {
 						return ret_value;
 					}
 
-					call_deferred(SNAME("_request_done"), RESULT_SUCCESS, response_code, response_headers, PackedByteArray());
+					_defer_done(RESULT_SUCCESS, response_code, response_headers, PackedByteArray());
 					return true;
 				}
 				if (body_len < 0) {
 					// Chunked transfer is done.
-					call_deferred(SNAME("_request_done"), RESULT_SUCCESS, response_code, response_headers, body);
+					_defer_done(RESULT_SUCCESS, response_code, response_headers, body);
 					return true;
 				}
 
-				call_deferred(SNAME("_request_done"), RESULT_CHUNKED_BODY_SIZE_MISMATCH, response_code, response_headers, PackedByteArray());
+				_defer_done(RESULT_CHUNKED_BODY_SIZE_MISMATCH, response_code, response_headers, PackedByteArray());
 				return true;
 				// Request might have been done.
 			} else {
@@ -324,7 +340,7 @@ bool HTTPRequest::_update_connection() {
 				int size = request_data.size();
 				Error err = client->request(method, request_string, headers, size > 0 ? request_data.ptr() : nullptr, size);
 				if (err != OK) {
-					call_deferred(SNAME("_request_done"), RESULT_CONNECTION_ERROR, 0, PackedStringArray(), PackedByteArray());
+					_defer_done(RESULT_CONNECTION_ERROR, 0, PackedStringArray(), PackedByteArray());
 					return true;
 				}
 
@@ -347,7 +363,7 @@ bool HTTPRequest::_update_connection() {
 				}
 
 				if (!client->is_response_chunked() && client->get_response_body_length() == 0) {
-					call_deferred(SNAME("_request_done"), RESULT_SUCCESS, response_code, response_headers, PackedByteArray());
+					_defer_done(RESULT_SUCCESS, response_code, response_headers, PackedByteArray());
 					return true;
 				}
 
@@ -356,14 +372,14 @@ bool HTTPRequest::_update_connection() {
 				body_len = client->get_response_body_length();
 
 				if (body_size_limit >= 0 && body_len > body_size_limit) {
-					call_deferred(SNAME("_request_done"), RESULT_BODY_SIZE_LIMIT_EXCEEDED, response_code, response_headers, PackedByteArray());
+					_defer_done(RESULT_BODY_SIZE_LIMIT_EXCEEDED, response_code, response_headers, PackedByteArray());
 					return true;
 				}
 
 				if (!download_to_file.is_empty()) {
 					file = FileAccess::open(download_to_file, FileAccess::WRITE);
 					if (file.is_null()) {
-						call_deferred(SNAME("_request_done"), RESULT_DOWNLOAD_FILE_CANT_OPEN, response_code, response_headers, PackedByteArray());
+						_defer_done(RESULT_DOWNLOAD_FILE_CANT_OPEN, response_code, response_headers, PackedByteArray());
 						return true;
 					}
 				}
@@ -375,14 +391,33 @@ bool HTTPRequest::_update_connection() {
 			}
 
 			PackedByteArray chunk = client->read_response_body_chunk();
+			downloaded.add(chunk.size());
+
+			// Decompress chunk if needed.
+			if (decompressor.is_valid()) {
+				Error err = decompressor->put_data(chunk.ptr(), chunk.size());
+				if (err == OK) {
+					chunk.resize(decompressor->get_available_bytes());
+					err = decompressor->get_data(chunk.ptrw(), chunk.size());
+				}
+				if (err != OK) {
+					_defer_done(RESULT_BODY_DECOMPRESS_FAILED, response_code, response_headers, PackedByteArray());
+					return true;
+				}
+			}
+			final_body_size.add(chunk.size());
+
+			if (body_size_limit >= 0 && final_body_size.get() > body_size_limit) {
+				_defer_done(RESULT_BODY_SIZE_LIMIT_EXCEEDED, response_code, response_headers, PackedByteArray());
+				return true;
+			}
 
 			if (chunk.size()) {
-				downloaded.add(chunk.size());
 				if (file.is_valid()) {
 					const uint8_t *r = chunk.ptr();
 					file->store_buffer(r, chunk.size());
 					if (file->get_error() != OK) {
-						call_deferred(SNAME("_request_done"), RESULT_DOWNLOAD_FILE_WRITE_ERROR, response_code, response_headers, PackedByteArray());
+						_defer_done(RESULT_DOWNLOAD_FILE_WRITE_ERROR, response_code, response_headers, PackedByteArray());
 						return true;
 					}
 				} else {
@@ -390,19 +425,14 @@ bool HTTPRequest::_update_connection() {
 				}
 			}
 
-			if (body_size_limit >= 0 && downloaded.get() > body_size_limit) {
-				call_deferred(SNAME("_request_done"), RESULT_BODY_SIZE_LIMIT_EXCEEDED, response_code, response_headers, PackedByteArray());
-				return true;
-			}
-
 			if (body_len >= 0) {
 				if (downloaded.get() == body_len) {
-					call_deferred(SNAME("_request_done"), RESULT_SUCCESS, response_code, response_headers, body);
+					_defer_done(RESULT_SUCCESS, response_code, response_headers, body);
 					return true;
 				}
 			} else if (client->get_status() == HTTPClient::STATUS_DISCONNECTED) {
 				// We read till EOF, with no errors. Request is done.
-				call_deferred(SNAME("_request_done"), RESULT_SUCCESS, response_code, response_headers, body);
+				_defer_done(RESULT_SUCCESS, response_code, response_headers, body);
 				return true;
 			}
 
@@ -410,11 +440,11 @@ bool HTTPRequest::_update_connection() {
 
 		} break; // Request resulted in body: break which must be read.
 		case HTTPClient::STATUS_CONNECTION_ERROR: {
-			call_deferred(SNAME("_request_done"), RESULT_CONNECTION_ERROR, 0, PackedStringArray(), PackedByteArray());
+			_defer_done(RESULT_CONNECTION_ERROR, 0, PackedStringArray(), PackedByteArray());
 			return true;
 		} break;
 		case HTTPClient::STATUS_TLS_HANDSHAKE_ERROR: {
-			call_deferred(SNAME("_request_done"), RESULT_TLS_HANDSHAKE_ERROR, 0, PackedStringArray(), PackedByteArray());
+			_defer_done(RESULT_TLS_HANDSHAKE_ERROR, 0, PackedStringArray(), PackedByteArray());
 			return true;
 		} break;
 	}
@@ -422,41 +452,13 @@ bool HTTPRequest::_update_connection() {
 	ERR_FAIL_V(false);
 }
 
+void HTTPRequest::_defer_done(int p_status, int p_code, const PackedStringArray &p_headers, const PackedByteArray &p_data) {
+	call_deferred(SNAME("_request_done"), p_status, p_code, p_headers, p_data);
+}
+
 void HTTPRequest::_request_done(int p_status, int p_code, const PackedStringArray &p_headers, const PackedByteArray &p_data) {
 	cancel_request();
 
-	// Determine if the request body is compressed.
-	bool is_compressed;
-	String content_encoding = get_header_value(p_headers, "Content-Encoding").to_lower();
-	Compression::Mode mode;
-	if (content_encoding == "gzip") {
-		mode = Compression::Mode::MODE_GZIP;
-		is_compressed = true;
-	} else if (content_encoding == "deflate") {
-		mode = Compression::Mode::MODE_DEFLATE;
-		is_compressed = true;
-	} else {
-		is_compressed = false;
-	}
-
-	if (accept_gzip && is_compressed && p_data.size() > 0) {
-		// Decompress request body
-		PackedByteArray decompressed;
-		int result = Compression::decompress_dynamic(&decompressed, body_size_limit, p_data.ptr(), p_data.size(), mode);
-		if (result == OK) {
-			emit_signal(SNAME("request_completed"), p_status, p_code, p_headers, decompressed);
-			return;
-		} else if (result == -5) {
-			WARN_PRINT("Decompressed size of HTTP response body exceeded body_size_limit");
-			p_status = RESULT_BODY_SIZE_LIMIT_EXCEEDED;
-			// Just return the raw data if we failed to decompress it.
-		} else {
-			WARN_PRINT("Failed to decompress HTTP response body");
-			p_status = RESULT_BODY_DECOMPRESS_FAILED;
-			// Just return the raw data if we failed to decompress it.
-		}
-	}
-
 	emit_signal(SNAME("request_completed"), p_status, p_code, p_headers, p_data);
 }
 
@@ -566,7 +568,7 @@ double HTTPRequest::get_timeout() {
 
 void HTTPRequest::_timeout() {
 	cancel_request();
-	call_deferred(SNAME("_request_done"), RESULT_TIMEOUT, 0, PackedStringArray(), PackedByteArray());
+	_defer_done(RESULT_TIMEOUT, 0, PackedStringArray(), PackedByteArray());
 }
 
 void HTTPRequest::_bind_methods() {
@@ -594,7 +596,6 @@ void HTTPRequest::_bind_methods() {
 	ClassDB::bind_method(D_METHOD("get_downloaded_bytes"), &HTTPRequest::get_downloaded_bytes);
 	ClassDB::bind_method(D_METHOD("get_body_size"), &HTTPRequest::get_body_size);
 
-	ClassDB::bind_method(D_METHOD("_redirect_request"), &HTTPRequest::_redirect_request);
 	ClassDB::bind_method(D_METHOD("_request_done"), &HTTPRequest::_request_done);
 
 	ClassDB::bind_method(D_METHOD("set_timeout", "timeout"), &HTTPRequest::set_timeout);
diff --git a/scene/main/http_request.h b/scene/main/http_request.h
index 290bacd9d2..80445684b0 100644
--- a/scene/main/http_request.h
+++ b/scene/main/http_request.h
@@ -32,6 +32,7 @@
 #define HTTP_REQUEST_H
 
 #include "core/io/http_client.h"
+#include "core/io/stream_peer_gzip.h"
 #include "core/os/thread.h"
 #include "core/templates/safe_refcount.h"
 #include "scene/main/node.h"
@@ -84,10 +85,12 @@ private:
 
 	String download_to_file;
 
+	Ref<StreamPeerGZIP> decompressor;
 	Ref<FileAccess> file;
 
 	int body_len = -1;
 	SafeNumeric<int> downloaded;
+	SafeNumeric<int> final_body_size;
 	int body_size_limit = -1;
 
 	int redirections = 0;
@@ -113,6 +116,7 @@ private:
 
 	Thread thread;
 
+	void _defer_done(int p_status, int p_code, const PackedStringArray &p_headers, const PackedByteArray &p_data);
 	void _request_done(int p_status, int p_code, const PackedStringArray &p_headers, const PackedByteArray &p_data);
 	static void _thread_func(void *p_userdata);