From 4a2da10e38c2b413c9e4db411e47d1b90d98d8ee Mon Sep 17 00:00:00 2001 From: AnotherTest Date: Sat, 26 Dec 2020 17:14:12 +0330 Subject: ProtocolServer: Stream the downloaded data if possible This patchset makes ProtocolServer stream the downloads to its client (LibProtocol), and as such changes the download API; a possible download lifecycle could be as such: notation = client->server:'>', server->client:'<', pipe activity:'*' ``` > StartDownload(GET, url, headers, {}) < Response(0, fd 8) * {data, 1024b} < HeadersBecameAvailable(0, response_headers, 200) < DownloadProgress(0, 4K, 1024) * {data, 1024b} * {data, 1024b} < DownloadProgress(0, 4K, 2048) * {data, 1024b} < DownloadProgress(0, 4K, 1024) < DownloadFinished(0, true, 4K) ``` Since managing the received file descriptor is a pain, LibProtocol implements `Download::stream_into(OutputStream)`, which can be used to stream the download into any given output stream (be it a file, or memory, or writing stuff with a delay, etc.). Also, as some of the users of this API require all the downloaded data upfront, LibProtocol also implements `set_should_buffer_all_input()`, which causes the download instance to buffer all the data until the download is complete, and to call the `on_buffered_download_finish` hook. --- Libraries/LibProtocol/Download.cpp | 88 ++++++++++++++++++++++++++++++++------ 1 file changed, 75 insertions(+), 13 deletions(-) (limited to 'Libraries/LibProtocol/Download.cpp') diff --git a/Libraries/LibProtocol/Download.cpp b/Libraries/LibProtocol/Download.cpp index b71127951b..f5011d04a5 100644 --- a/Libraries/LibProtocol/Download.cpp +++ b/Libraries/LibProtocol/Download.cpp @@ -41,25 +41,81 @@ bool Download::stop() return m_client->stop_download({}, *this); } -void Download::did_finish(Badge, bool success, Optional status_code, u32 total_size, i32 shbuf_id, const IPC::Dictionary& response_headers) +void Download::stream_into(OutputStream& stream) { - if (!on_finish) + ASSERT(!m_internal_stream_data); + + auto notifier = Core::Notifier::construct(fd(), Core::Notifier::Read); + + m_internal_stream_data = make(fd()); + m_internal_stream_data->read_notifier = notifier; + + auto user_on_finish = move(on_finish); + on_finish = [this](auto success, auto total_size) { + m_internal_stream_data->success = success; + m_internal_stream_data->total_size = total_size; + m_internal_stream_data->download_done = true; + }; + + notifier->on_ready_to_read = [this, &stream, user_on_finish = move(user_on_finish)] { + constexpr size_t buffer_size = 1 * KiB; + static char buf[buffer_size]; + auto nread = m_internal_stream_data->read_stream.read({ buf, buffer_size }); + if (!stream.write_or_error({ buf, nread })) { + // FIXME: What do we do here? + TODO(); + } + + if (m_internal_stream_data->read_stream.eof() || (m_internal_stream_data->download_done && !m_internal_stream_data->success)) { + m_internal_stream_data->read_notifier->close(); + user_on_finish(m_internal_stream_data->success, m_internal_stream_data->total_size); + } else { + m_internal_stream_data->read_stream.handle_any_error(); + } + }; +} + +void Download::set_should_buffer_all_input(bool value) +{ + if (m_should_buffer_all_input == value) return; - ReadonlyBytes payload; - RefPtr shared_buffer; - if (success && shbuf_id != -1) { - shared_buffer = SharedBuffer::create_from_shbuf_id(shbuf_id); - payload = { shared_buffer->data(), total_size }; + if (m_internal_buffered_data && !value) { + m_internal_buffered_data = nullptr; + m_should_buffer_all_input = false; + return; } - // FIXME: It's a bit silly that we copy the response headers here just so we can move them into a HashMap with different traits. - HashMap caseless_response_headers; - response_headers.for_each_entry([&](auto& name, auto& value) { - caseless_response_headers.set(name, value); - }); + ASSERT(!m_internal_stream_data); + ASSERT(!m_internal_buffered_data); + ASSERT(on_buffered_download_finish); // Not having this set makes no sense. + m_internal_buffered_data = make(fd()); + m_should_buffer_all_input = true; - on_finish(success, payload, move(shared_buffer), caseless_response_headers, status_code); + on_headers_received = [this](auto& headers, auto response_code) { + m_internal_buffered_data->response_headers = headers; + m_internal_buffered_data->response_code = move(response_code); + }; + + on_finish = [this](auto success, u32 total_size) { + auto output_buffer = m_internal_buffered_data->payload_stream.copy_into_contiguous_buffer(); + on_buffered_download_finish( + success, + total_size, + m_internal_buffered_data->response_headers, + m_internal_buffered_data->response_code, + output_buffer); + }; + + stream_into(m_internal_buffered_data->payload_stream); +} + +void Download::did_finish(Badge, bool success, u32 total_size) +{ + if (!on_finish) + return; + + on_finish(success, total_size); } void Download::did_progress(Badge, Optional total_size, u32 downloaded_size) @@ -68,6 +124,12 @@ void Download::did_progress(Badge, Optional total_size, u32 downloa on_progress(total_size, downloaded_size); } +void Download::did_receive_headers(Badge, const HashMap& response_headers, Optional response_code) +{ + if (on_headers_received) + on_headers_received(response_headers, response_code); +} + void Download::did_request_certificates(Badge) { if (on_certificate_requested) { -- cgit v1.2.3