From 9124c68278014aa362ba9518cd647e861c57dd2f Mon Sep 17 00:00:00 2001 From: nerix Date: Sun, 9 Mar 2025 21:32:49 +0100 Subject: [PATCH] fix(eventsub): check keepalive (#6058) --- CHANGELOG.md | 2 +- .../benchmarks/src/parse.cpp | 6 +- .../payloads/session-welcome.hpp | 3 +- .../include/twitch-eventsub-ws/session.hpp | 37 ++- .../generated/payloads/session-welcome.cpp | 20 ++ lib/twitch-eventsub-ws/src/session.cpp | 287 ++++++++++-------- lib/twitch-eventsub-ws/tests/src/parse.cpp | 6 +- tests/src/EventSubMessages.cpp | 11 +- 8 files changed, 224 insertions(+), 148 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea907d89..7a4f904c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,7 +45,7 @@ - Bugfix: Fixed a thick border on Windows 11. (#5836) - Bugfix: Fixed some windows not immediately closing. (#6054) - Dev: Subscriptions to PubSub channel points redemption topics now use no auth token, making it continue to work during PubSub shutdown. (#5947) -- Dev: Add initial experimental EventSub support. (#5837, #5895, #5897, #5904, #5910, #5903, #5915, #5916, #5930, #5935, #5932, #5943, #5952, #5953, #5968, #5973, #5974, #5980, #5981, #5985, #5990, #5992, #5993, #5996, #5995, #6000, #6001, #6002, #6003, #6005, #6007, #6010, #6008, #6012, #6013, #6015, #6017, #6027, #6028, #6035, #6036, #6040, #6041, #6048) +- Dev: Add initial experimental EventSub support. (#5837, #5895, #5897, #5904, #5910, #5903, #5915, #5916, #5930, #5935, #5932, #5943, #5952, #5953, #5968, #5973, #5974, #5980, #5981, #5985, #5990, #5992, #5993, #5996, #5995, #6000, #6001, #6002, #6003, #6005, #6007, #6010, #6008, #6012, #6013, #6015, #6017, #6027, #6028, #6035, #6036, #6040, #6041, #6048, #6058) - Dev: Remove unneeded platform specifier for toasts. (#5914) - Dev: Cleanly shutdown on `SIGINT`/`SIGTERM` on Linux & macOS. (#6053) - Dev: Highlight checks now use non-capturing groups for the boundaries. (#5784) diff --git a/lib/twitch-eventsub-ws/benchmarks/src/parse.cpp b/lib/twitch-eventsub-ws/benchmarks/src/parse.cpp index 240934fe..730fa179 100644 --- a/lib/twitch-eventsub-ws/benchmarks/src/parse.cpp +++ b/lib/twitch-eventsub-ws/benchmarks/src/parse.cpp @@ -171,12 +171,16 @@ void BM_ParseAndHandleMessages(benchmark::State &state) auto messages = readMessages(); std::unique_ptr listener = std::make_unique(); + boost::asio::io_context ioc; + boost::asio::ssl::context ssl( + boost::asio::ssl::context::method::tls_client); + Session sess(ioc, ssl, std::move(listener)); for (auto _ : state) { for (const auto &msg : messages) { - boost::system::error_code ec = handleMessage(listener, msg); + boost::system::error_code ec = sess.handleMessage(msg); assert(!ec); } } diff --git a/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/payloads/session-welcome.hpp b/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/payloads/session-welcome.hpp index b6ec7ea4..89c7369e 100644 --- a/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/payloads/session-welcome.hpp +++ b/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/payloads/session-welcome.hpp @@ -24,7 +24,8 @@ namespace chatterino::eventsub::lib::payload::session_welcome { /// json_inner=session struct Payload { std::string id; - std::optional reconnectURL; + std::optional reconnectURL; // null in welcome message + std::optional keepaliveTimeoutSeconds; // null in reconnect message }; #include "twitch-eventsub-ws/payloads/session-welcome.inc" diff --git a/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/session.hpp b/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/session.hpp index d03c7b07..957b4b81 100644 --- a/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/session.hpp +++ b/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/session.hpp @@ -7,21 +7,16 @@ #include #include +namespace chatterino::eventsub::lib::messages { + +struct Metadata; + +} // namespace chatterino::eventsub::lib::messages + namespace chatterino::eventsub::lib { class Listener; -/** - * handleMessage takes the incoming message in the buffer, parses it - * as JSON then forwards it to the listener, if applicable. - * - * This is called from the Session, and is only provided if you are interested - * in building your own boost asio framework thing - **/ -boost::system::error_code handleMessage( - std::unique_ptr &listener, - const boost::beast::flat_buffer &buffer); - class Session : public std::enable_shared_from_this { public: @@ -38,13 +33,17 @@ public: Listener *getListener(); + // public for testing + boost::system::error_code handleMessage( + const boost::beast::flat_buffer &buffer); + private: void onResolve(boost::beast::error_code ec, - boost::asio::ip::tcp::resolver::results_type results); + const boost::asio::ip::tcp::resolver::results_type &results); void onConnect( boost::beast::error_code ec, - boost::asio::ip::tcp::resolver::results_type::endpoint_type ep); + const boost::asio::ip::tcp::resolver::results_type::endpoint_type &ep); void onSSLHandshake(boost::beast::error_code ec); @@ -56,6 +55,14 @@ private: void fail(boost::beast::error_code ec, std::string_view op); + boost::system::error_code onSessionWelcome( + const messages::Metadata &metadata, const boost::json::value &jv); + boost::system::error_code onSessionReconnect(const boost::json::value &jv); + boost::system::error_code onNotification(const messages::Metadata &metadata, + const boost::json::value &jv); + + void checkKeepalive(); + boost::asio::ip::tcp::resolver resolver; boost::beast::websocket::stream< boost::beast::ssl_stream> @@ -66,6 +73,10 @@ private: std::string path; std::string userAgent; std::unique_ptr listener; + + std::chrono::seconds keepaliveTimeout{0}; + bool receivedMessage = false; + std::unique_ptr keepaliveTimer; }; } // namespace chatterino::eventsub::lib diff --git a/lib/twitch-eventsub-ws/src/generated/payloads/session-welcome.cpp b/lib/twitch-eventsub-ws/src/generated/payloads/session-welcome.cpp index c1b10e26..dd9cd84c 100644 --- a/lib/twitch-eventsub-ws/src/generated/payloads/session-welcome.cpp +++ b/lib/twitch-eventsub-ws/src/generated/payloads/session-welcome.cpp @@ -56,9 +56,29 @@ boost::json::result_for::type tag_invoke( reconnectURL = std::move(treconnectURL.value()); } + static_assert( + std::is_trivially_copyable_v().keepaliveTimeoutSeconds)>>); + std::optional keepaliveTimeoutSeconds = std::nullopt; + const auto *jvkeepaliveTimeoutSeconds = + root.if_contains("keepalive_timeout_seconds"); + if (jvkeepaliveTimeoutSeconds != nullptr && + !jvkeepaliveTimeoutSeconds->is_null()) + { + auto tkeepaliveTimeoutSeconds = + boost::json::try_value_to(*jvkeepaliveTimeoutSeconds); + + if (tkeepaliveTimeoutSeconds.has_error()) + { + return tkeepaliveTimeoutSeconds.error(); + } + keepaliveTimeoutSeconds = tkeepaliveTimeoutSeconds.value(); + } + return Payload{ .id = std::move(id.value()), .reconnectURL = std::move(reconnectURL), + .keepaliveTimeoutSeconds = keepaliveTimeoutSeconds, }; } diff --git a/lib/twitch-eventsub-ws/src/session.cpp b/lib/twitch-eventsub-ws/src/session.cpp index b480874f..13d5a3ee 100644 --- a/lib/twitch-eventsub-ws/src/session.cpp +++ b/lib/twitch-eventsub-ws/src/session.cpp @@ -14,11 +14,9 @@ #include #include -#include #include #include #include -#include #include namespace beast = boost::beast; @@ -37,12 +35,6 @@ using NotificationHandlers = std::unordered_map< std::unique_ptr &)>, boost::hash>; -using MessageHandlers = std::unordered_map< - std::string, - std::function &, const NotificationHandlers &)>>; - namespace { template @@ -234,124 +226,8 @@ namespace { // Add your new subscription types above this line }; - const MessageHandlers MESSAGE_HANDLERS{ - { - "session_welcome", - [](const auto &metadata, const auto &jv, auto &listener, - const auto & /*notificationHandlers*/) { - auto oPayload = - parsePayload(jv); - if (!oPayload) - { - // TODO: error handling - return oPayload.error(); - } - const auto &payload = *oPayload; - - listener->onSessionWelcome(metadata, payload); - return boost::system::error_code{}; - }, - }, - { - "session_keepalive", - [](const auto &metadata, const auto &jv, auto &listener, - const auto ¬ificationHandlers) { - // TODO: should we do something here? - return boost::system::error_code{}; - }, - }, - {"session_reconnect", - [](const auto & /*metadata*/, const auto &jv, auto &listener, - const auto & /*notificationHandlers*/) { - auto oPayload = - parsePayload(jv); - if (!oPayload) - { - return oPayload.error(); - } - const auto &payload = *oPayload; - auto *listenerPtr = listener.get(); - - listenerPtr->onClose(std::move(listener), payload.reconnectURL); - return boost::system::error_code{}; - }}, - { - "notification", - [](const auto &metadata, const auto &jv, auto &listener, - const auto ¬ificationHandlers) { - listener->onNotification(metadata, jv); - - if (!metadata.subscriptionType || !metadata.subscriptionVersion) - { - // TODO: error handling - return boost::system::error_code{}; - } - - auto it = - notificationHandlers.find({*metadata.subscriptionType, - *metadata.subscriptionVersion}); - if (it == notificationHandlers.end()) - { - EVENTSUB_BAIL_HERE(error::Kind::NoMessageHandler); - } - - return it->second(metadata, jv, listener); - }, - }, - }; - } // namespace -boost::system::error_code handleMessage(std::unique_ptr &listener, - const beast::flat_buffer &buffer) -{ - boost::system::error_code parseError; - auto jv = - boost::json::parse(beast::buffers_to_string(buffer.data()), parseError); - if (parseError) - { - // TODO: wrap error? - return parseError; - } - - const auto *jvObject = jv.if_object(); - if (jvObject == nullptr) - { - EVENTSUB_BAIL_HERE(error::Kind::ExpectedObject); - } - - const auto *metadataV = jvObject->if_contains("metadata"); - if (metadataV == nullptr) - { - EVENTSUB_BAIL_HERE(error::Kind::FieldMissing); - } - auto metadataResult = - boost::json::try_value_to(*metadataV); - if (metadataResult.has_error()) - { - // TODO: wrap error? - return metadataResult.error(); - } - - const auto &metadata = metadataResult.value(); - - auto handler = MESSAGE_HANDLERS.find(metadata.messageType); - - if (handler == MESSAGE_HANDLERS.end()) - { - EVENTSUB_BAIL_HERE(error::Kind::NoMessageHandler); - } - - const auto *payloadV = jvObject->if_contains("payload"); - if (payloadV == nullptr) - { - EVENTSUB_BAIL_HERE(error::Kind::FieldMissing); - } - - return handler->second(metadata, *payloadV, listener, - NOTIFICATION_HANDLERS); -} - // Resolver and socket require an io_context Session::Session(boost::asio::io_context &ioc, boost::asio::ssl::context &ctx, std::unique_ptr listener) @@ -392,8 +268,9 @@ Listener *Session::getListener() return this->listener.get(); } -void Session::onResolve(beast::error_code ec, - boost::asio::ip::tcp::resolver::results_type results) +void Session::onResolve( + beast::error_code ec, + const boost::asio::ip::tcp::resolver::results_type &results) { if (ec) { @@ -412,7 +289,7 @@ void Session::onResolve(beast::error_code ec, void Session::onConnect( beast::error_code ec, - boost::asio::ip::tcp::resolver::results_type::endpoint_type ep) + const boost::asio::ip::tcp::resolver::results_type::endpoint_type &ep) { if (ec) { @@ -500,7 +377,8 @@ void Session::onRead(beast::error_code ec, std::size_t bytes_transferred) return; } - auto messageError = handleMessage(this->listener, this->buffer); + this->receivedMessage = true; + auto messageError = this->handleMessage(this->buffer); if (messageError) { this->fail(messageError, "handleMessage"); @@ -543,8 +421,161 @@ void Session::fail(beast::error_code ec, std::string_view op) std::cerr << op << ": " << ec.message() << " (" << ec.location() << ")\n"; if (!this->ws.is_open() && this->listener) { + if (this->keepaliveTimer) + { + this->keepaliveTimer.reset(); + } + this->listener->onClose(std::move(this->listener), {}); } } +boost::system::error_code Session::handleMessage( + const beast::flat_buffer &buffer) +{ + boost::system::error_code parseError; + auto jv = + boost::json::parse(beast::buffers_to_string(buffer.data()), parseError); + if (parseError) + { + // TODO: wrap error? + return parseError; + } + + const auto *jvObject = jv.if_object(); + if (jvObject == nullptr) + { + EVENTSUB_BAIL_HERE(error::Kind::ExpectedObject); + } + + const auto *metadataV = jvObject->if_contains("metadata"); + if (metadataV == nullptr) + { + EVENTSUB_BAIL_HERE(error::Kind::FieldMissing); + } + auto metadataResult = + boost::json::try_value_to(*metadataV); + if (metadataResult.has_error()) + { + // TODO: wrap error? + return metadataResult.error(); + } + + const auto &metadata = metadataResult.value(); + + const auto *payloadV = jvObject->if_contains("payload"); + if (payloadV == nullptr) + { + EVENTSUB_BAIL_HERE(error::Kind::FieldMissing); + } + + if (metadata.messageType == "notification") + { + return this->onNotification(metadata, *payloadV); + } + if (metadata.messageType == "session_welcome") + { + return this->onSessionWelcome(metadata, *payloadV); + } + if (metadata.messageType == "session_keepalive") + { + return {}; // nothing to do + } + if (metadata.messageType == "session_reconnect") + { + return this->onSessionReconnect(*payloadV); + } + EVENTSUB_BAIL_HERE(error::Kind::NoMessageHandler); +} + +boost::system::error_code Session::onSessionWelcome( + const messages::Metadata &metadata, const boost::json::value &jv) +{ + auto oPayload = parsePayload(jv); + if (!oPayload) + { + // TODO: error handling + return oPayload.error(); + } + const auto &payload = *oPayload; + + listener->onSessionWelcome(metadata, payload); + + // we're graceful with the keepalive timeout + this->keepaliveTimeout = + std::chrono::seconds{payload.keepaliveTimeoutSeconds.value_or(60)} * 2; + assert(!this->keepaliveTimer); + std::cerr << "Keepalive: " << this->keepaliveTimeout.count() << 's'; + this->checkKeepalive(); + + return {}; +} + +boost::system::error_code Session::onSessionReconnect( + const boost::json::value &jv) +{ + auto oPayload = parsePayload(jv); + if (!oPayload) + { + return oPayload.error(); + } + const auto &payload = *oPayload; + auto *listenerPtr = listener.get(); + + listenerPtr->onClose(std::move(listener), payload.reconnectURL); + return {}; +} + +boost::system::error_code Session::onNotification( + const messages::Metadata &metadata, const boost::json::value &jv) +{ + listener->onNotification(metadata, jv); + + if (!metadata.subscriptionType || !metadata.subscriptionVersion) + { + // TODO: error handling + return boost::system::error_code{}; + } + + auto it = NOTIFICATION_HANDLERS.find( + {*metadata.subscriptionType, *metadata.subscriptionVersion}); + if (it == NOTIFICATION_HANDLERS.end()) + { + EVENTSUB_BAIL_HERE(error::Kind::NoMessageHandler); + } + + return it->second(metadata, jv, listener); +} + +void Session::checkKeepalive() +{ + if (!this->receivedMessage) + { + std::cerr << "Keepalive timeout, closing\n"; + if (this->listener) + { + this->listener->onClose(std::move(this->listener), {}); + } + return; + } + this->receivedMessage = false; + + if (this->keepaliveTimeout.count() == 0) + { + return; + } + + this->keepaliveTimer = + std::make_unique(this->ws.get_executor()); + this->keepaliveTimer->expires_after(this->keepaliveTimeout); + this->keepaliveTimer->async_wait([this](boost::system::error_code ec) { + if (ec) + { + std::cerr << "Keepalive timer cancelled: " << ec.message() << '\n'; + return; + } + this->checkKeepalive(); + }); +} + } // namespace chatterino::eventsub::lib diff --git a/lib/twitch-eventsub-ws/tests/src/parse.cpp b/lib/twitch-eventsub-ws/tests/src/parse.cpp index c792cec2..ce7096e3 100644 --- a/lib/twitch-eventsub-ws/tests/src/parse.cpp +++ b/lib/twitch-eventsub-ws/tests/src/parse.cpp @@ -154,7 +154,11 @@ TEST_P(TestHandleMessageP, Run) auto buf = readToFlatBuffer(filePath(GetParam() + ".json")); std::unique_ptr listener = std::make_unique(); - auto ec = handleMessage(listener, buf); + boost::asio::io_context ioc; + boost::asio::ssl::context ssl( + boost::asio::ssl::context::method::tls_client); + Session sess(ioc, ssl, std::move(listener)); + auto ec = sess.handleMessage(buf); ASSERT_FALSE(ec.failed()) << ec.what() << ec.message() << ec.location().to_string(); } diff --git a/tests/src/EventSubMessages.cpp b/tests/src/EventSubMessages.cpp index 723dc75c..ee2b0ad7 100644 --- a/tests/src/EventSubMessages.cpp +++ b/tests/src/EventSubMessages.cpp @@ -304,6 +304,13 @@ TEST_P(TestEventSubMessagesP, Run) input = snapshot->input().toArray(); } + std::unique_ptr listener = + std::make_unique(); + boost::asio::io_context ioc; + boost::asio::ssl::context ssl( + boost::asio::ssl::context::method::tls_client); + eventsub::lib::Session sess(ioc, ssl, std::move(listener)); + for (const auto inputRef : input) { auto inputObj = inputRef.toObject(); @@ -321,9 +328,7 @@ TEST_P(TestEventSubMessagesP, Run) auto json = makePayload(eventSubscription->second, inputObj); - std::unique_ptr listener = - std::make_unique(); - auto ec = eventsub::lib::handleMessage(listener, json); + auto ec = sess.handleMessage(json); ASSERT_FALSE(ec.failed()) << ec.what() << ec.message() << ec.location().to_string(); }