From 914257d537ae3e0b0238bb52e0346fb1d6e794d3 Mon Sep 17 00:00:00 2001 From: nerix Date: Thu, 6 Mar 2025 22:41:35 +0100 Subject: [PATCH] feat(eventsub): reconnect (#6035) --- CHANGELOG.md | 2 +- .../include/twitch-eventsub-ws/listener.hpp | 4 + .../payloads/session-welcome.hpp | 1 + .../include/twitch-eventsub-ws/session.hpp | 25 ++-- .../generated/payloads/session-welcome.cpp | 15 +++ lib/twitch-eventsub-ws/src/session.cpp | 70 +++++++--- src/Application.cpp | 1 - src/common/Args.cpp | 16 +++ src/common/Args.hpp | 5 + src/common/network/NetworkPrivate.hpp | 3 + src/common/network/NetworkRequest.cpp | 8 ++ src/common/network/NetworkRequest.hpp | 4 + src/common/network/NetworkTask.cpp | 10 ++ src/providers/twitch/api/Helix.cpp | 19 ++- src/providers/twitch/eventsub/Connection.cpp | 21 ++- src/providers/twitch/eventsub/Connection.hpp | 5 +- src/providers/twitch/eventsub/Controller.cpp | 121 ++++++++++++++++-- src/providers/twitch/eventsub/Controller.hpp | 20 +++ 18 files changed, 305 insertions(+), 45 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bec44ce9..dd19617f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,7 +37,7 @@ - Bugfix: Fixed color input thinking blue is also red. (#5982) - Bugfix: Fixed an issue where commands would sometimes reset if Chatterino was improperly shut down. (#6011) - 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) +- 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) - Dev: Remove unneeded platform specifier for toasts. (#5914) - Dev: Highlight checks now use non-capturing groups for the boundaries. (#5784) - Dev: Removed unused PubSub whisper code. (#5898) diff --git a/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/listener.hpp b/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/listener.hpp index 5a4f207b..8039eb43 100644 --- a/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/listener.hpp +++ b/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/listener.hpp @@ -30,6 +30,10 @@ public: virtual void onNotification(const messages::Metadata &metadata, const boost::json::value &jv) = 0; + virtual void onClose( + std::unique_ptr self, + const std::optional & /* reconnectUrl */) {}; + // Subscription types virtual void onChannelBan( const messages::Metadata &metadata, 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 b71391cb..b6ec7ea4 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,6 +24,7 @@ namespace chatterino::eventsub::lib::payload::session_welcome { /// json_inner=session struct Payload { std::string id; + std::optional reconnectURL; }; #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 6d8ce052..d03c7b07 100644 --- a/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/session.hpp +++ b/lib/twitch-eventsub-ws/include/twitch-eventsub-ws/session.hpp @@ -22,20 +22,8 @@ boost::system::error_code handleMessage( std::unique_ptr &listener, const boost::beast::flat_buffer &buffer); -// Sends a WebSocket message and prints the response class Session : public std::enable_shared_from_this { - boost::asio::ip::tcp::resolver resolver; - boost::beast::websocket::stream< - boost::beast::ssl_stream> - ws; - boost::beast::flat_buffer buffer; - std::string host; - std::string port; - std::string path; - std::string userAgent; - std::unique_ptr listener; - public: // Resolver and socket require an io_context explicit Session(boost::asio::io_context &ioc, @@ -65,6 +53,19 @@ private: void onRead(boost::beast::error_code ec, std::size_t bytes_transferred); void onClose(boost::beast::error_code ec); + + void fail(boost::beast::error_code ec, std::string_view op); + + boost::asio::ip::tcp::resolver resolver; + boost::beast::websocket::stream< + boost::beast::ssl_stream> + ws; + boost::beast::flat_buffer buffer; + std::string host; + std::string port; + std::string path; + std::string userAgent; + std::unique_ptr listener; }; } // 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 4347b6e4..c1b10e26 100644 --- a/lib/twitch-eventsub-ws/src/generated/payloads/session-welcome.cpp +++ b/lib/twitch-eventsub-ws/src/generated/payloads/session-welcome.cpp @@ -42,8 +42,23 @@ boost::json::result_for::type tag_invoke( return id.error(); } + std::optional reconnectURL = std::nullopt; + const auto *jvreconnectURL = root.if_contains("reconnect_url"); + if (jvreconnectURL != nullptr && !jvreconnectURL->is_null()) + { + auto treconnectURL = + boost::json::try_value_to(*jvreconnectURL); + + if (treconnectURL.has_error()) + { + return treconnectURL.error(); + } + reconnectURL = std::move(treconnectURL.value()); + } + return Payload{ .id = std::move(id.value()), + .reconnectURL = std::move(reconnectURL), }; } diff --git a/lib/twitch-eventsub-ws/src/session.cpp b/lib/twitch-eventsub-ws/src/session.cpp index 9ab7c729..b480874f 100644 --- a/lib/twitch-eventsub-ws/src/session.cpp +++ b/lib/twitch-eventsub-ws/src/session.cpp @@ -45,13 +45,6 @@ using MessageHandlers = std::unordered_map< namespace { - // Report a failure - void fail(beast::error_code ec, char const *what) - { - std::cerr << what << ": " << ec.message() << " (" << ec.location() - << ")\n"; - } - template boost::system::result parsePayload(const boost::json::value &jv) { @@ -267,6 +260,21 @@ namespace { 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, @@ -389,7 +397,8 @@ void Session::onResolve(beast::error_code ec, { if (ec) { - return fail(ec, "resolve"); + this->fail(ec, "resolve"); + return; } // Set a timeout on the operation @@ -407,7 +416,8 @@ void Session::onConnect( { if (ec) { - return fail(ec, "connect"); + this->fail(ec, "connect"); + return; } // Set a timeout on the operation @@ -419,7 +429,8 @@ void Session::onConnect( { ec = beast::error_code(static_cast(::ERR_get_error()), boost::asio::error::get_ssl_category()); - return fail(ec, "connect"); + this->fail(ec, "connect"); + return; } // Update the host_ string. This will provide the value of the @@ -438,7 +449,8 @@ void Session::onSSLHandshake(beast::error_code ec) { if (ec) { - return fail(ec, "ssl_handshake"); + this->fail(ec, "ssl_handshake"); + return; } // Turn off the timeout on the tcp_stream, because @@ -465,7 +477,8 @@ void Session::onHandshake(beast::error_code ec) { if (ec) { - return fail(ec, "handshake"); + this->fail(ec, "handshake"); + return; } this->ws.async_read(buffer, beast::bind_front_handler(&Session::onRead, @@ -476,19 +489,31 @@ void Session::onRead(beast::error_code ec, std::size_t bytes_transferred) { boost::ignore_unused(bytes_transferred); + if (!this->listener) + { + return; + } + if (ec) { - return fail(ec, "read"); + this->fail(ec, "read"); + return; } auto messageError = handleMessage(this->listener, this->buffer); if (messageError) { - fail(messageError, "handleMessage"); + this->fail(messageError, "handleMessage"); } this->buffer.clear(); + if (!this->listener) + { + this->close(); + return; + } + this->ws.async_read(buffer, beast::bind_front_handler(&Session::onRead, shared_from_this())); } @@ -502,13 +527,24 @@ void Session::onClose(beast::error_code ec) { if (ec) { - return fail(ec, "close"); + this->fail(ec, "close"); + return; } // If we get here then the connection is closed gracefully + if (this->listener) + { + this->listener->onClose(std::move(this->listener), {}); + } +} - // The make_printable() function helps print a ConstBufferSequence - std::cout << beast::make_printable(buffer.data()) << std::endl; +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) + { + this->listener->onClose(std::move(this->listener), {}); + } } } // namespace chatterino::eventsub::lib diff --git a/src/Application.cpp b/src/Application.cpp index 2838bb29..8b545b1a 100644 --- a/src/Application.cpp +++ b/src/Application.cpp @@ -592,7 +592,6 @@ pronouns::Pronouns *Application::getPronouns() eventsub::IController *Application::getEventSub() { - assertInGuiThread(); assert(this->eventSub); return this->eventSub.get(); diff --git a/src/common/Args.cpp b/src/common/Args.cpp index 795e1fa1..6558ce3b 100644 --- a/src/common/Args.cpp +++ b/src/common/Args.cpp @@ -140,6 +140,12 @@ Args::Args(const QApplication &app, const Paths &paths) "specified, Twitch is assumed.", "t:channel"); +#ifndef NDEBUG + QCommandLineOption useLocalEventsubOption( + "use-local-eventsub", + "Use the local eventsub server at 127.0.0.1:3012."); +#endif + parser.addOptions({ {{"V", "version"}, "Displays version information."}, crashRecoveryOption, @@ -152,6 +158,9 @@ Args::Args(const QApplication &app, const Paths &paths) loginOption, channelLayout, activateOption, +#ifndef NDEBUG + useLocalEventsubOption, +#endif }); if (!parser.parse(app.arguments())) @@ -216,6 +225,13 @@ Args::Args(const QApplication &app, const Paths &paths) parseActivateOption(parser.value(activateOption)); } +#ifndef NDEBUG + if (parser.isSet(useLocalEventsubOption)) + { + this->useLocalEventsub = true; + } +#endif + this->currentArguments_ = extractCommandLine(parser, { verboseOption, safeModeOption, diff --git a/src/common/Args.hpp b/src/common/Args.hpp index cac8fb67..a1689603 100644 --- a/src/common/Args.hpp +++ b/src/common/Args.hpp @@ -64,6 +64,11 @@ public: bool verbose{}; bool safeMode{}; +#ifndef NDEBUG + // twitch event websocket start-server --ssl --port 3012 + bool useLocalEventsub = false; +#endif + QStringList currentArguments() const; private: diff --git a/src/common/network/NetworkPrivate.hpp b/src/common/network/NetworkPrivate.hpp index f39efe69..a1688a72 100644 --- a/src/common/network/NetworkPrivate.hpp +++ b/src/common/network/NetworkPrivate.hpp @@ -53,6 +53,9 @@ public: /// By default, there's no explicit timeout for the request. /// To set a timeout, use NetworkRequest's timeout method std::optional timeout{}; +#ifndef NDEBUG + bool ignoreSslErrors = false; // for local eventsub +#endif QString getHash(); diff --git a/src/common/network/NetworkRequest.cpp b/src/common/network/NetworkRequest.cpp index 9413943b..34d5604e 100644 --- a/src/common/network/NetworkRequest.cpp +++ b/src/common/network/NetworkRequest.cpp @@ -215,4 +215,12 @@ NetworkRequest NetworkRequest::json(const QByteArray &payload) && .header("Accept", "application/json"); } +#ifndef NDEBUG +NetworkRequest NetworkRequest::ignoreSslErrors(bool ignore) && +{ + this->data->ignoreSslErrors = ignore; + return std::move(*this); +} +#endif + } // namespace chatterino diff --git a/src/common/network/NetworkRequest.hpp b/src/common/network/NetworkRequest.hpp index 1308fb02..50515f66 100644 --- a/src/common/network/NetworkRequest.hpp +++ b/src/common/network/NetworkRequest.hpp @@ -76,6 +76,10 @@ public: NetworkRequest json(const QJsonDocument &document) &&; NetworkRequest json(const QByteArray &payload) &&; +#ifndef NDEBUG + NetworkRequest ignoreSslErrors(bool ignore) &&; +#endif + void execute(); private: diff --git a/src/common/network/NetworkTask.cpp b/src/common/network/NetworkTask.cpp index 7bc1330c..c2640a0b 100644 --- a/src/common/network/NetworkTask.cpp +++ b/src/common/network/NetworkTask.cpp @@ -61,6 +61,16 @@ void NetworkTask::run() QObject::connect(this->reply_, &QNetworkReply::finished, this, &NetworkTask::finished); + +#ifndef NDEBUG + if (this->data_->ignoreSslErrors) + { + QObject::connect(this->reply_, &QNetworkReply::sslErrors, this, + [this](const auto &errors) { + this->reply_->ignoreSslErrors(errors); + }); + } +#endif } QNetworkReply *NetworkTask::createReply() diff --git a/src/providers/twitch/api/Helix.cpp b/src/providers/twitch/api/Helix.cpp index 3ececeda..ca92588b 100644 --- a/src/providers/twitch/api/Helix.cpp +++ b/src/providers/twitch/api/Helix.cpp @@ -1,5 +1,7 @@ #include "providers/twitch/api/Helix.hpp" +#include "Application.hpp" +#include "common/Args.hpp" #include "common/Literals.hpp" #include "common/network/NetworkRequest.hpp" #include "common/network/NetworkResult.hpp" @@ -3393,7 +3395,16 @@ NetworkRequest Helix::makeRequest(const QString &url, const QUrlQuery &urlQuery, // return std::nullopt; } - const QString baseUrl("https://api.twitch.tv/helix/"); + QString baseUrl("https://api.twitch.tv/helix/"); + +#ifndef NDEBUG + bool ignoreSslErrors = + getApp()->getArgs().useLocalEventsub && url == "eventsub/subscriptions"; + if (ignoreSslErrors) + { + baseUrl = "https://127.0.0.1:3012/"; + } +#endif QUrl fullUrl(baseUrl + url); @@ -3403,7 +3414,11 @@ NetworkRequest Helix::makeRequest(const QString &url, const QUrlQuery &urlQuery, .timeout(5 * 1000) .header("Accept", "application/json") .header("Client-ID", this->clientId) - .header("Authorization", "Bearer " + this->oauthToken); + .header("Authorization", "Bearer " + this->oauthToken) +#ifndef NDEBUG + .ignoreSslErrors(ignoreSslErrors) +#endif + ; } NetworkRequest Helix::makeGet(const QString &url, const QUrlQuery &urlQuery) diff --git a/src/providers/twitch/eventsub/Connection.cpp b/src/providers/twitch/eventsub/Connection.cpp index 422096b3..e81f196d 100644 --- a/src/providers/twitch/eventsub/Connection.cpp +++ b/src/providers/twitch/eventsub/Connection.cpp @@ -6,6 +6,7 @@ #include "controllers/highlights/HighlightController.hpp" #include "messages/Message.hpp" #include "messages/MessageBuilder.hpp" +#include "providers/twitch/eventsub/Controller.hpp" #include "providers/twitch/eventsub/MessageBuilder.hpp" #include "providers/twitch/eventsub/MessageHandlers.hpp" #include "providers/twitch/PubSubActions.hpp" @@ -18,7 +19,7 @@ #include "util/PostToThread.hpp" #include -#include +#include #include #include @@ -71,6 +72,19 @@ void Connection::onNotification(const lib::messages::Metadata &metadata, qCDebug(LOG) << "on notification: " << jsonString.c_str(); } +void Connection::onClose(std::unique_ptr self, + const std::optional &reconnectURL) +{ + auto *app = tryGetApp(); + if (!app) + { + return; + } + + app->getEventSub()->reconnectConnection(std::move(self), reconnectURL, + this->subscriptions); +} + void Connection::onChannelBan( const lib::messages::Metadata &metadata, const lib::payload::channel_ban::v1::Payload &payload) @@ -387,4 +401,9 @@ void Connection::markRequestSubscribed(const SubscriptionRequest &request) this->subscriptions.emplace(request); } +void Connection::markRequestUnsubscribed(const SubscriptionRequest &request) +{ + this->subscriptions.erase(request); +} + } // namespace chatterino::eventsub diff --git a/src/providers/twitch/eventsub/Connection.hpp b/src/providers/twitch/eventsub/Connection.hpp index 6aad8dd2..04ec28f7 100644 --- a/src/providers/twitch/eventsub/Connection.hpp +++ b/src/providers/twitch/eventsub/Connection.hpp @@ -21,6 +21,9 @@ public: void onNotification(const lib::messages::Metadata &metadata, const boost::json::value &jv) override; + void onClose(std::unique_ptr self, + const std::optional &reconnectURL) override; + void onChannelBan( const lib::messages::Metadata &metadata, const lib::payload::channel_ban::v1::Payload &payload) override; @@ -85,7 +88,7 @@ public: bool isSubscribedTo(const SubscriptionRequest &request) const; void markRequestSubscribed(const SubscriptionRequest &request); - // TODO: Add an "markRequestUnsubscribed" method + void markRequestUnsubscribed(const SubscriptionRequest &request); private: QString sessionID; diff --git a/src/providers/twitch/eventsub/Controller.cpp b/src/providers/twitch/eventsub/Controller.cpp index f7c141c9..b5958711 100644 --- a/src/providers/twitch/eventsub/Controller.cpp +++ b/src/providers/twitch/eventsub/Controller.cpp @@ -1,5 +1,7 @@ #include "providers/twitch/eventsub/Controller.hpp" +#include "Application.hpp" +#include "common/Args.hpp" #include "common/QLogging.hpp" #include "common/Version.hpp" #include "providers/twitch/api/Helix.hpp" @@ -17,16 +19,16 @@ namespace { -/// Enable LOCAL_EVENTSUB when you want to debug eventsub with a local instance of the Twitch CLI -/// twitch event websocket start-server --ssl --port 3012 -constexpr bool LOCAL_EVENTSUB = false; +using namespace chatterino; std::tuple getEventSubHost() { - if constexpr (LOCAL_EVENTSUB) +#ifndef NDEBUG + if (getApp()->getArgs().useLocalEventsub) { return {"localhost", "3012", "/ws"}; } +#endif return {"eventsub.wss.twitch.tv", "443", "/ws"}; } @@ -222,6 +224,54 @@ SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request) return handle; } +void Controller::reconnectConnection( + std::unique_ptr connection, + const std::optional &reconnectURL, + const std::unordered_set &subs) +{ + this->clearConnections(); + if (subs.empty()) + { + return; + } + + if (reconnectURL) + { + qCDebug(chatterinoTwitchEventSub) << "Using reconnect URL to reconnect"; + // this is epic + QUrl url(QString::fromStdString(*reconnectURL)); + this->createConnection(url.host(QUrl::FullyEncoded).toStdString(), + std::to_string(url.port(443)), + url.path(QUrl::FullyEncoded).toStdString(), + std::move(connection)); + return; + } + + // no reconnect URL - something happened + // but first, clear the subscriptions + qCDebug(chatterinoTwitchEventSub) + << "Resubscribing to topics after connection failure"; + { + std::lock_guard g(this->subscriptionsMutex); + for (const auto &sub : subs) + { + auto it = this->subscriptions.find(sub); + if (it != this->subscriptions.end()) + { + qCDebug(chatterinoTwitchEventSub) << "Resetting" << it->first; + it->second.connection = {}; + it->second.retryAttempts = 0; + it->second.state = Subscription::State::Subscribing; + } + } + } + + for (const auto &sub : subs) + { + this->subscribe(sub, false); + } +} + void Controller::subscribe(const SubscriptionRequest &request, bool isRetry) { // 1. Flush dead connections (maybe this should not be done here) @@ -355,8 +405,10 @@ std::optional> Controller::getViableConnection( auto *listener = dynamic_cast(connection->getListener()); - assert(listener != nullptr && "Something goofy has gone wrong, Session " - "listener must be our Connection type"); + if (!listener) + { + continue; // dead connection + } if (listener->getSessionID().isEmpty()) { @@ -374,13 +426,23 @@ std::optional> Controller::getViableConnection( } void Controller::createConnection() +{ + this->createConnection(this->eventSubHost, this->eventSubPort, + this->eventSubPath, std::make_unique()); +} + +void Controller::createConnection(std::string host, std::string port, + std::string path, + std::unique_ptr listener) { try { boost::asio::ssl::context sslContext{ boost::asio::ssl::context::tlsv12_client}; - if constexpr (!LOCAL_EVENTSUB) +#ifndef NDEBUG + if (!getApp()->getArgs().useLocalEventsub) +#endif { sslContext.set_verify_mode( boost::asio::ssl::verify_peer | @@ -391,12 +453,12 @@ void Controller::createConnection() } auto connection = std::make_shared( - this->ioContext, sslContext, std::make_unique()); + this->ioContext, sslContext, std::move(listener)); this->registerConnection(connection); - connection->run(this->eventSubHost, this->eventSubPort, - this->eventSubPath, this->userAgent); + connection->run(std::move(host), std::move(port), std::move(path), + this->userAgent); } catch (std::exception &e) { @@ -509,6 +571,20 @@ void Controller::markRequestSubscribed(const SubscriptionRequest &request, std::lock_guard lock(this->subscriptionsMutex); + auto strong = connection.lock(); + if (!strong) + { + // we disconnected + return; + } + auto *listener = dynamic_cast(strong->getListener()); + if (!listener) + { + // we disconnected + return; + } + listener->markRequestSubscribed(request); + auto &subscription = this->subscriptions[request]; assert((subscription.state == Subscription::State::Subscribing || @@ -560,6 +636,31 @@ void Controller::markRequestUnsubscribed(const SubscriptionRequest &request) qCDebug(LOG) << "Set state to unsubscribed" << request; subscription.state = Subscription::State::Unsubscribed; subscription.retryAttempts = 0; + auto conn = subscription.connection.lock(); + if (conn) + { + auto *listener = dynamic_cast(conn->getListener()); + if (listener) + { + listener->markRequestUnsubscribed(request); + } + } + subscription.connection = {}; +} + +void Controller::clearConnections() +{ + std::erase_if(this->connections, [](const auto &it) { + auto conn = it.lock(); + return !conn || !conn->getListener(); + }); +} + +void DummyController::reconnectConnection( + std::unique_ptr /* connection */, + const std::optional & /* reconnectURL */, + const std::unordered_set & /* subs */) +{ } } // namespace chatterino::eventsub diff --git a/src/providers/twitch/eventsub/Controller.hpp b/src/providers/twitch/eventsub/Controller.hpp index fd6bd817..2916cbc4 100644 --- a/src/providers/twitch/eventsub/Controller.hpp +++ b/src/providers/twitch/eventsub/Controller.hpp @@ -17,6 +17,7 @@ #include #include #include +#include namespace chatterino::eventsub { @@ -44,6 +45,11 @@ public: /// create a new connection and queue up the subscription to run again after X seconds. [[nodiscard]] virtual SubscriptionHandle subscribe( const SubscriptionRequest &request) = 0; + + virtual void reconnectConnection( + std::unique_ptr connection, + const std::optional &reconnectURL, + const std::unordered_set &subs) = 0; }; class Controller : public IController @@ -59,10 +65,17 @@ public: [[nodiscard]] SubscriptionHandle subscribe( const SubscriptionRequest &request) override; + void reconnectConnection( + std::unique_ptr connection, + const std::optional &reconnectURL, + const std::unordered_set &subs) override; + private: void subscribe(const SubscriptionRequest &request, bool isRetry); void createConnection(); + void createConnection(std::string host, std::string port, std::string path, + std::unique_ptr listener); void registerConnection(std::weak_ptr &&connection); void retrySubscription(const SubscriptionRequest &request, @@ -77,6 +90,8 @@ private: void markRequestUnsubscribed(const SubscriptionRequest &request); + void clearConnections(); + const std::string userAgent; std::string eventSubHost; @@ -154,6 +169,11 @@ public: (void)request; return {}; } + + void reconnectConnection( + std::unique_ptr connection, + const std::optional &reconnectURL, + const std::unordered_set &subs) override; }; } // namespace chatterino::eventsub