diff options
| -rw-r--r-- | include/AsyncTask.hpp | 65 | ||||
| -rw-r--r-- | src/QuickMedia.cpp | 66 |
2 files changed, 97 insertions, 34 deletions
diff --git a/include/AsyncTask.hpp b/include/AsyncTask.hpp new file mode 100644 index 0000000..0d9c453 --- /dev/null +++ b/include/AsyncTask.hpp @@ -0,0 +1,65 @@ +#pragma once + +#include "Program.hpp" +#include <thread> +#include <future> + +namespace QuickMedia { + template <typename T> + class AsyncTask { + public: + using CallbackFunc = std::function<T()>; + + AsyncTask() = default; + + AsyncTask(CallbackFunc callback_func) { + std::promise<T> promise; + future = promise.get_future(); + thread = std::thread(&AsyncTask::thread_handler, this, std::move(promise), std::move(callback_func)); + } + + AsyncTask& operator=(CallbackFunc callback_func) { + cancel(); + std::promise<T> promise; + future = promise.get_future(); + thread = std::thread(&AsyncTask::thread_handler, this, std::move(promise), std::move(callback_func)); + return *this; + } + + ~AsyncTask() { + cancel(); + } + + bool valid() { + return future.valid(); + } + + bool ready() { + return future.valid() && future.wait_for(std::chrono::seconds(0)) == std::future_status::ready; + } + + T get() { + thread.join(); + return future.get(); + } + + void cancel() { + if(valid()) { + program_kill_in_thread(thread.get_id()); + get(); + } + } + private: + void thread_handler(std::promise<T> &&promise, CallbackFunc callback_func) { + if constexpr(std::is_same<T, void>::value) { + callback_func(); + promise.set_value(); + } else { + promise.set_value(callback_func()); + } + } + private: + std::thread thread; + std::future<T> future; + }; +}
\ No newline at end of file diff --git a/src/QuickMedia.cpp b/src/QuickMedia.cpp index 3c76079..0347cbb 100644 --- a/src/QuickMedia.cpp +++ b/src/QuickMedia.cpp @@ -21,6 +21,7 @@ #include "../include/NetUtils.hpp" #include "../include/SfmlFixes.hpp" #include "../include/FontLoader.hpp" +#include "../include/AsyncTask.hpp" #include "../external/hash-library/sha256.h" #include <assert.h> @@ -1518,6 +1519,7 @@ namespace QuickMedia { if(event.type == sf::Event::Closed) { program_kill_in_thread(task_thread.get_id()); task_thread.join(); + future.get(); current_page = PageType::EXIT; window.close(); return TaskResult::CANCEL; @@ -1529,14 +1531,14 @@ namespace QuickMedia { } else if(event.type == sf::Event::KeyPressed && event.key.code == sf::Keyboard::Escape) { program_kill_in_thread(task_thread.get_id()); task_thread.join(); + future.get(); return TaskResult::CANCEL; } } if(is_future_ready(future)) { - bool future_result = future.get(); task_thread.join(); - if(!future_result) + if(!future.get()) return TaskResult::FALSE; break; } @@ -3122,7 +3124,7 @@ namespace QuickMedia { const sf::Int32 read_marker_timeout_ms_default = 3000; sf::Int32 read_marker_timeout_ms = 0; - std::future<void> set_read_marker_future; + AsyncTask<void> set_read_marker_future; bool setting_read_marker = false; bool redraw = true; @@ -3288,7 +3290,7 @@ namespace QuickMedia { if(chat_state == ChatState::TYPING_MESSAGE) { post_task_queue.push([this, ¤t_room, text, msgtype]() { if(matrix->post_message(current_room, text, std::nullopt, std::nullopt, msgtype) != PluginResult::OK) - show_notification("QuickMedia", "Failed to post matrix message", Urgency::CRITICAL); + fprintf(stderr, "Failed to post matrix message\n"); }); chat_input.set_editable(false); chat_state = ChatState::NAVIGATING; @@ -3297,7 +3299,7 @@ namespace QuickMedia { void *related_to_message = currently_operating_on_item->userdata; post_task_queue.push([this, ¤t_room, text, related_to_message]() { if(matrix->post_reply(current_room, text, related_to_message) != PluginResult::OK) - show_notification("QuickMedia", "Failed to post matrix reply", Urgency::CRITICAL); + fprintf(stderr, "Failed to post matrix reply\n"); }); chat_input.set_editable(false); chat_state = ChatState::NAVIGATING; @@ -3307,7 +3309,7 @@ namespace QuickMedia { void *related_to_message = currently_operating_on_item->userdata; post_task_queue.push([this, ¤t_room, text, related_to_message]() { if(matrix->post_edit(current_room, text, related_to_message) != PluginResult::OK) - show_notification("QuickMedia", "Failed to post matrix edit", Urgency::CRITICAL); + fprintf(stderr, "Failed to post matrix edit\n"); }); chat_input.set_editable(false); chat_state = ChatState::NAVIGATING; @@ -3318,12 +3320,12 @@ namespace QuickMedia { return false; }; - std::future<Messages> previous_messages_future; + AsyncTask<Messages> previous_messages_future; bool fetching_previous_messages_running = false; RoomData *previous_messages_future_room = nullptr; //const int num_fetch_message_threads = 4; - std::future<std::shared_ptr<Message>> fetch_message_future; + AsyncTask<std::shared_ptr<Message>> fetch_message_future; bool fetching_message_running = false; RoomData *fetch_future_room = nullptr; BodyItem *fetch_body_item = nullptr; @@ -3356,9 +3358,9 @@ namespace QuickMedia { event_data->status = FetchStatus::LOADING; fetch_message_tab = PINNED_TAB_INDEX; // TODO: Check if the message is already cached before calling async? is this needed? is async creation expensive? - fetch_message_future = std::async(std::launch::async, [this, &fetch_future_room, message_event_id]() { + fetch_message_future = [this, &fetch_future_room, message_event_id]() { return matrix->get_message_by_id(fetch_future_room, message_event_id); - }); + }; }; // TODO: How about instead fetching all messages we have, not only the visible ones? also fetch with multiple threads. @@ -3387,9 +3389,9 @@ namespace QuickMedia { body_item->embedded_item_status = FetchStatus::LOADING; fetch_message_tab = MESSAGES_TAB_INDEX; // TODO: Check if the message is already cached before calling async? is this needed? is async creation expensive? - fetch_message_future = std::async(std::launch::async, [this, &fetch_future_room, message_event_id]() { + fetch_message_future = [this, &fetch_future_room, message_event_id]() { return matrix->get_message_by_id(fetch_future_room, message_event_id); - }); + }; }; const float tab_spacer_height = 0.0f; @@ -3582,12 +3584,12 @@ namespace QuickMedia { gradient_inc = 0; fetching_previous_messages_running = true; previous_messages_future_room = current_room; - previous_messages_future = std::async(std::launch::async, [this, &previous_messages_future_room]() { + previous_messages_future = [this, &previous_messages_future_room]() { Messages messages; if(matrix->get_previous_room_messages(previous_messages_future_room, messages) != PluginResult::OK) fprintf(stderr, "Failed to get previous matrix messages in room: %s\n", previous_messages_future_room->id.c_str()); return messages; - }); + }; } } else if(event.key.code == sf::Keyboard::Down) { tabs[selected_tab].body->select_next_item(); @@ -3714,7 +3716,7 @@ namespace QuickMedia { std::string err_msg; if(matrix->delete_message(current_room, selected_message, err_msg) != PluginResult::OK) { // TODO: Show inline notification - show_notification("QuickMedia", "Failed to delete message, reason: " + err_msg, Urgency::CRITICAL); + fprintf(stderr, "Failed to delete message, reason: %s\n", err_msg.c_str()); } }); } else { @@ -3831,17 +3833,15 @@ namespace QuickMedia { break; } case PageType::CHAT_LOGIN: { - // TODO: Cancel these instead - if(set_read_marker_future.valid()) - set_read_marker_future.get(); - if(previous_messages_future.valid()) - previous_messages_future.get(); - if(fetch_message_future.valid()) - fetch_message_future.get(); + set_read_marker_future.cancel(); + previous_messages_future.cancel(); + fetch_message_future.cancel(); typing_state_queue.close(); + program_kill_in_thread(typing_state_thread.get_id()); if(typing_state_thread.joinable()) typing_state_thread.join(); post_task_queue.close(); + program_kill_in_thread(post_thread.get_id()); if(post_thread.joinable()) post_thread.join(); new_page = PageType::CHAT; @@ -3948,13 +3948,13 @@ namespace QuickMedia { modify_related_messages_in_current_room(sync_data.messages); process_pinned_events(sync_data.pinned_events); - if(is_future_ready(set_read_marker_future)) { + if(set_read_marker_future.ready()) { set_read_marker_future.get(); read_marker_timer.restart(); setting_read_marker = false; } - if(fetching_previous_messages_running && is_future_ready(previous_messages_future)) { + if(fetching_previous_messages_running && previous_messages_future.ready()) { Messages new_messages = previous_messages_future.get(); fprintf(stderr, "Finished fetching older messages, num new messages: %zu\n", new_messages.size()); // Ignore finished fetch of messages if it happened in another room. When we navigate back to the room we will get the messages again @@ -3975,7 +3975,7 @@ namespace QuickMedia { fetching_previous_messages_running = false; } - if(fetching_message_running && is_future_ready(fetch_message_future)) { + if(fetching_message_running && fetch_message_future.ready()) { std::shared_ptr<Message> message = fetch_message_future.get(); fprintf(stderr, "Finished fetching message: %s\n", message ? message->event_id.c_str() : "(null)"); // Ignore finished fetch of messages if it happened in another room. When we navigate back to the room we will get the messages again @@ -4126,11 +4126,11 @@ namespace QuickMedia { // TODO: What if the message is no longer valid? setting_read_marker = true; RoomData *room = current_room; - set_read_marker_future = std::async(std::launch::async, [this, room, message]() mutable { + set_read_marker_future = [this, room, message]() mutable { if(matrix->set_read_marker(room, message) != PluginResult::OK) { fprintf(stderr, "Warning: failed to set read marker to %s\n", message->event_id.c_str()); } - }); + }; } } @@ -4143,17 +4143,15 @@ namespace QuickMedia { } chat_page_end: - // TODO: Cancel these instead - if(set_read_marker_future.valid()) - set_read_marker_future.get(); - if(previous_messages_future.valid()) - previous_messages_future.get(); - if(fetch_message_future.valid()) - fetch_message_future.get(); + set_read_marker_future.cancel(); + previous_messages_future.cancel(); + fetch_message_future.cancel(); typing_state_queue.close(); + program_kill_in_thread(typing_state_thread.get_id()); if(typing_state_thread.joinable()) typing_state_thread.join(); post_task_queue.close(); + program_kill_in_thread(post_thread.get_id()); if(post_thread.joinable()) post_thread.join(); |