From 7f393e2401519016a7dba6cfb2e1f76a89645e00 Mon Sep 17 00:00:00 2001 From: nerix Date: Sat, 8 Nov 2025 13:46:10 +0100 Subject: [PATCH] refactor(liveupdates): use `WebSocketPool` over websocketpp (#6308) Reviewed-by: pajlada --- CHANGELOG.md | 1 + src/CMakeLists.txt | 4 +- src/common/websockets/WebSocketPool.cpp | 7 +- src/common/websockets/WebSocketPool.hpp | 4 +- .../websockets/detail/WebSocketPoolImpl.cpp | 13 +- .../websockets/detail/WebSocketPoolImpl.hpp | 3 +- src/providers/bttv/BttvLiveUpdates.cpp | 123 +++-- src/providers/bttv/BttvLiveUpdates.hpp | 41 +- .../bttv/liveupdates/BttvLiveUpdateClient.cpp | 78 ++++ .../bttv/liveupdates/BttvLiveUpdateClient.hpp | 22 + .../liveupdates/BasicPubSubClient.hpp | 127 ++--- .../liveupdates/BasicPubSubListener.hpp | 66 +++ .../liveupdates/BasicPubSubManager.hpp | 385 +++++---------- .../liveupdates/BasicPubSubWebsocket.hpp | 36 -- src/providers/liveupdates/Diag.hpp | 13 + src/providers/seventv/SeventvEventAPI.cpp | 441 ++++-------------- src/providers/seventv/SeventvEventAPI.hpp | 60 +-- src/providers/seventv/eventapi/Client.cpp | 360 ++++++++++++-- src/providers/seventv/eventapi/Client.hpp | 39 +- src/providers/seventv/eventapi/Message.cpp | 4 +- src/providers/seventv/eventapi/Message.hpp | 2 +- src/providers/twitch/TwitchIrcServer.cpp | 5 +- tests/src/BasicPubSub.cpp | 59 ++- tests/src/BttvLiveUpdates.cpp | 26 +- tests/src/IrcMessageHandler.cpp | 1 + tests/src/SeventvEventAPI.cpp | 37 +- 26 files changed, 984 insertions(+), 973 deletions(-) create mode 100644 src/providers/bttv/liveupdates/BttvLiveUpdateClient.cpp create mode 100644 src/providers/bttv/liveupdates/BttvLiveUpdateClient.hpp create mode 100644 src/providers/liveupdates/BasicPubSubListener.hpp delete mode 100644 src/providers/liveupdates/BasicPubSubWebsocket.hpp create mode 100644 src/providers/liveupdates/Diag.hpp diff --git a/CHANGELOG.md b/CHANGELOG.md index e961e446..560838cb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,7 @@ - Dev: Fix Arch Linux partial upgrade error in CI. (#6553) - Dev: Added Qt keyword and warning flags project wide. (#6520) - Def: Fixed compilation error in tests with Clang 21. (#6519) +- Dev: The 7TV and BTTV liveupdates now use Boost.Beast's WebSockets. (#6308) - Dev: Fixed compilation warnings on clang-cl. (#6528) - Dev: Fixed compilation error in tests with Clang 21. (#6519) - Dev: Use CMake's `FetchContent` for RapidJSON, PajladaSignals, PajladaSerialize, and PajladaSettings. (#6560, #6567, #6569) diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 739eaf58..9923391a 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -343,6 +343,8 @@ set(SOURCE_FILES providers/bttv/BttvLiveUpdates.cpp providers/bttv/BttvLiveUpdates.hpp + providers/bttv/liveupdates/BttvLiveUpdateClient.cpp + providers/bttv/liveupdates/BttvLiveUpdateClient.hpp providers/bttv/liveupdates/BttvLiveUpdateMessages.cpp providers/bttv/liveupdates/BttvLiveUpdateMessages.hpp providers/bttv/liveupdates/BttvLiveUpdateSubscription.cpp @@ -374,8 +376,8 @@ set(SOURCE_FILES providers/links/LinkResolver.hpp providers/liveupdates/BasicPubSubClient.hpp + providers/liveupdates/BasicPubSubListener.hpp providers/liveupdates/BasicPubSubManager.hpp - providers/liveupdates/BasicPubSubWebsocket.hpp providers/pronouns/Pronouns.cpp providers/pronouns/Pronouns.hpp diff --git a/src/common/websockets/WebSocketPool.cpp b/src/common/websockets/WebSocketPool.cpp index 6e2e5301..9d583a24 100644 --- a/src/common/websockets/WebSocketPool.cpp +++ b/src/common/websockets/WebSocketPool.cpp @@ -6,7 +6,9 @@ namespace chatterino { -WebSocketPool::WebSocketPool() = default; +WebSocketPool::WebSocketPool(QString shortName) + : shortName(std::move(shortName)) {}; + WebSocketPool::~WebSocketPool() { if (this->impl) @@ -34,7 +36,8 @@ WebSocketHandle WebSocketPool::createSocket( { try { - this->impl = std::make_unique(); + this->impl = std::make_unique( + this->shortName); } catch (const boost::system::system_error &err) { diff --git a/src/common/websockets/WebSocketPool.hpp b/src/common/websockets/WebSocketPool.hpp index f5349d45..9986475b 100644 --- a/src/common/websockets/WebSocketPool.hpp +++ b/src/common/websockets/WebSocketPool.hpp @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include @@ -74,7 +75,7 @@ struct WebSocketOptions { class WebSocketPool { public: - WebSocketPool(); + WebSocketPool(QString shortName = {}); ~WebSocketPool(); [[nodiscard]] WebSocketHandle createSocket( @@ -82,6 +83,7 @@ public: private: std::unique_ptr impl; + QString shortName; }; } // namespace chatterino diff --git a/src/common/websockets/detail/WebSocketPoolImpl.cpp b/src/common/websockets/detail/WebSocketPoolImpl.cpp index c1ed3090..ced880b7 100644 --- a/src/common/websockets/detail/WebSocketPoolImpl.cpp +++ b/src/common/websockets/detail/WebSocketPoolImpl.cpp @@ -6,10 +6,11 @@ #include "util/RenameThread.hpp" #include +#include namespace chatterino::ws::detail { -WebSocketPoolImpl::WebSocketPoolImpl() +WebSocketPoolImpl::WebSocketPoolImpl(const QString &shortName) : ioc(1) , ssl(boost::asio::ssl::context::tls_client) , work(this->ioc.get_executor()) @@ -43,7 +44,15 @@ WebSocketPoolImpl::WebSocketPoolImpl() this->ioc.run(); this->shutdownFlag.set(); }); - renameThread(*this->ioThread, "WebSocketPool"); + + auto threadName = [&]() -> QString { + if (shortName.isEmpty()) + { + return "WebSocketPool"; + } + return "WS-" % shortName; + }(); + renameThread(*this->ioThread, threadName); } WebSocketPoolImpl::~WebSocketPoolImpl() diff --git a/src/common/websockets/detail/WebSocketPoolImpl.hpp b/src/common/websockets/detail/WebSocketPoolImpl.hpp index 3d99346b..a7af7c51 100644 --- a/src/common/websockets/detail/WebSocketPoolImpl.hpp +++ b/src/common/websockets/detail/WebSocketPoolImpl.hpp @@ -4,6 +4,7 @@ #include #include +#include #include #include @@ -17,7 +18,7 @@ class WebSocketConnection; class WebSocketPoolImpl { public: - WebSocketPoolImpl(); + WebSocketPoolImpl(const QString &shortName); ~WebSocketPoolImpl(); WebSocketPoolImpl(const WebSocketPoolImpl &) = delete; diff --git a/src/providers/bttv/BttvLiveUpdates.cpp b/src/providers/bttv/BttvLiveUpdates.cpp index c248629b..172d0ab9 100644 --- a/src/providers/bttv/BttvLiveUpdates.cpp +++ b/src/providers/bttv/BttvLiveUpdates.cpp @@ -1,105 +1,90 @@ #include "providers/bttv/BttvLiveUpdates.hpp" -#include "common/Literals.hpp" +#include "liveupdates/BttvLiveUpdateClient.hpp" +#include "providers/liveupdates/BasicPubSubManager.hpp" #include +#include #include namespace chatterino { -using namespace chatterino::literals; +using namespace Qt::StringLiterals; -BttvLiveUpdates::BttvLiveUpdates(QString host) +class BttvLiveUpdatesPrivate + : public BasicPubSubManager +{ +public: + BttvLiveUpdatesPrivate(BttvLiveUpdates &parent, QString host); + ~BttvLiveUpdatesPrivate() override; + BttvLiveUpdatesPrivate(const BttvLiveUpdatesPrivate &) = delete; + BttvLiveUpdatesPrivate(const BttvLiveUpdatesPrivate &&) = delete; + BttvLiveUpdatesPrivate &operator=(const BttvLiveUpdatesPrivate &) = delete; + BttvLiveUpdatesPrivate &operator=(const BttvLiveUpdatesPrivate &&) = delete; + + std::shared_ptr makeClient(); + + // Contains all joined Twitch channel-ids + std::unordered_set joinedChannels; + BttvLiveUpdates &parent; + + friend BasicPubSubManager; + friend BttvLiveUpdates; +}; + +BttvLiveUpdatesPrivate::BttvLiveUpdatesPrivate(BttvLiveUpdates &parent, + QString host) : BasicPubSubManager(std::move(host), u"BTTV"_s) + , parent(parent) { } -BttvLiveUpdates::~BttvLiveUpdates() +BttvLiveUpdatesPrivate::~BttvLiveUpdatesPrivate() { this->stop(); } +std::shared_ptr BttvLiveUpdatesPrivate::makeClient() +{ + return std::make_shared(this->parent); +} + +BttvLiveUpdates::BttvLiveUpdates(QString host) + : private_(std::make_unique(*this, std::move(host))) +{ +} + +BttvLiveUpdates::~BttvLiveUpdates() = default; + void BttvLiveUpdates::joinChannel(const QString &channelID, const QString &userID) { - if (this->joinedChannels_.insert(channelID).second) + if (this->private_->joinedChannels.insert(channelID).second) { - this->subscribe({BttvLiveUpdateSubscriptionChannel{channelID}}); - this->subscribe({BttvLiveUpdateBroadcastMe{.twitchID = channelID, - .userID = userID}}); + this->private_->subscribe( + {BttvLiveUpdateSubscriptionChannel{channelID}}); + this->private_->subscribe({BttvLiveUpdateBroadcastMe{ + .twitchID = channelID, .userID = userID}}); } } void BttvLiveUpdates::partChannel(const QString &id) { - if (this->joinedChannels_.erase(id) > 0) + if (this->private_->joinedChannels.erase(id) > 0) { - this->unsubscribe({BttvLiveUpdateSubscriptionChannel{id}}); + this->private_->unsubscribe({BttvLiveUpdateSubscriptionChannel{id}}); } } -void BttvLiveUpdates::onMessage( - websocketpp::connection_hdl /*hdl*/, - BasicPubSubManager::WebsocketMessagePtr msg) +void BttvLiveUpdates::stop() { - const auto &payload = QString::fromStdString(msg->get_payload()); - QJsonDocument jsonDoc(QJsonDocument::fromJson(payload.toUtf8())); + this->private_->stop(); +} - if (jsonDoc.isNull()) - { - qCDebug(chatterinoBttv) << "Failed to parse live update JSON"; - return; - } - auto json = jsonDoc.object(); - - auto eventType = json["name"].toString(); - auto eventData = json["data"].toObject(); - - if (eventType == "emote_create") - { - auto message = BttvLiveUpdateEmoteUpdateAddMessage(eventData); - - if (!message.validate()) - { - qCDebug(chatterinoBttv) << "Invalid add message" << json; - return; - } - - this->signals_.emoteAdded.invoke(message); - } - else if (eventType == "emote_update") - { - auto message = BttvLiveUpdateEmoteUpdateAddMessage(eventData); - - if (!message.validate()) - { - qCDebug(chatterinoBttv) << "Invalid update message" << json; - return; - } - - this->signals_.emoteUpdated.invoke(message); - } - else if (eventType == "emote_delete") - { - auto message = BttvLiveUpdateEmoteRemoveMessage(eventData); - - if (!message.validate()) - { - qCDebug(chatterinoBttv) << "Invalid deletion message" << json; - return; - } - - this->signals_.emoteRemoved.invoke(message); - } - else if (eventType == "lookup_user") - { - // ignored - } - else - { - qCDebug(chatterinoBttv) << "Unhandled event:" << json; - } +const liveupdates::Diag &BttvLiveUpdates::diag() const +{ + return this->private_->diag; } } // namespace chatterino diff --git a/src/providers/bttv/BttvLiveUpdates.hpp b/src/providers/bttv/BttvLiveUpdates.hpp index 7861a505..7c9cd0c6 100644 --- a/src/providers/bttv/BttvLiveUpdates.hpp +++ b/src/providers/bttv/BttvLiveUpdates.hpp @@ -1,25 +1,28 @@ #pragma once -#include "providers/bttv/liveupdates/BttvLiveUpdateMessages.hpp" -#include "providers/bttv/liveupdates/BttvLiveUpdateSubscription.hpp" -#include "providers/liveupdates/BasicPubSubManager.hpp" -#include "util/QStringHash.hpp" - #include +#include -#include +#include namespace chatterino { -class BttvLiveUpdates : public BasicPubSubManager +namespace liveupdates { +struct Diag; +} // namespace liveupdates + +struct BttvLiveUpdateEmoteUpdateAddMessage; +struct BttvLiveUpdateEmoteRemoveMessage; + +class BttvLiveUpdatesPrivate; +class BttvLiveUpdates { template - using Signal = - pajlada::Signals::Signal; // type-id is vector> + using Signal = pajlada::Signals::Signal; public: BttvLiveUpdates(QString host); - ~BttvLiveUpdates() override; + ~BttvLiveUpdates(); struct { Signal emoteAdded; @@ -44,15 +47,19 @@ public: */ void partChannel(const QString &id); -protected: - void onMessage( - websocketpp::connection_hdl hdl, - BasicPubSubManager::WebsocketMessagePtr msg) - override; + /// Stop the manager + /// + /// Used in tests to check that connections are closed (through #diag()). + /// Otherwise, calling the destructor is sufficient. + void stop(); + + /// Statistics about the opened/closed connections and received messages + /// + /// Used in tests. + const liveupdates::Diag &diag() const; private: - // Contains all joined Twitch channel-ids - std::unordered_set joinedChannels_; + std::unique_ptr private_; }; } // namespace chatterino diff --git a/src/providers/bttv/liveupdates/BttvLiveUpdateClient.cpp b/src/providers/bttv/liveupdates/BttvLiveUpdateClient.cpp new file mode 100644 index 00000000..41d517cd --- /dev/null +++ b/src/providers/bttv/liveupdates/BttvLiveUpdateClient.cpp @@ -0,0 +1,78 @@ +#include "providers/bttv/liveupdates/BttvLiveUpdateClient.hpp" + +#include "providers/bttv/BttvLiveUpdates.hpp" +#include "providers/bttv/liveupdates/BttvLiveUpdateMessages.hpp" + +#include +#include +#include + +namespace chatterino { + +BttvLiveUpdateClient::BttvLiveUpdateClient(BttvLiveUpdates &manager) + : BasicPubSubClient(100) + , manager(manager) +{ +} + +void BttvLiveUpdateClient::onMessage(const QByteArray &msg) +{ + QJsonDocument jsonDoc(QJsonDocument::fromJson(msg)); + + if (jsonDoc.isNull()) + { + qCDebug(chatterinoBttv) << "Failed to parse live update JSON"; + return; + } + auto json = jsonDoc.object(); + + auto eventType = json["name"].toString(); + auto eventData = json["data"].toObject(); + + if (eventType == "emote_create") + { + auto message = BttvLiveUpdateEmoteUpdateAddMessage(eventData); + + if (!message.validate()) + { + qCDebug(chatterinoBttv) << "Invalid add message" << json; + return; + } + + this->manager.signals_.emoteAdded.invoke(message); + } + else if (eventType == "emote_update") + { + auto message = BttvLiveUpdateEmoteUpdateAddMessage(eventData); + + if (!message.validate()) + { + qCDebug(chatterinoBttv) << "Invalid update message" << json; + return; + } + + this->manager.signals_.emoteUpdated.invoke(message); + } + else if (eventType == "emote_delete") + { + auto message = BttvLiveUpdateEmoteRemoveMessage(eventData); + + if (!message.validate()) + { + qCDebug(chatterinoBttv) << "Invalid deletion message" << json; + return; + } + + this->manager.signals_.emoteRemoved.invoke(message); + } + else if (eventType == "lookup_user") + { + // ignored + } + else + { + qCDebug(chatterinoBttv) << "Unhandled event:" << json; + } +} + +} // namespace chatterino diff --git a/src/providers/bttv/liveupdates/BttvLiveUpdateClient.hpp b/src/providers/bttv/liveupdates/BttvLiveUpdateClient.hpp new file mode 100644 index 00000000..84902cb0 --- /dev/null +++ b/src/providers/bttv/liveupdates/BttvLiveUpdateClient.hpp @@ -0,0 +1,22 @@ +#pragma once + +#include "providers/bttv/liveupdates/BttvLiveUpdateSubscription.hpp" +#include "providers/liveupdates/BasicPubSubClient.hpp" + +namespace chatterino { + +class BttvLiveUpdates; + +class BttvLiveUpdateClient + : public BasicPubSubClient +{ +public: + BttvLiveUpdateClient(BttvLiveUpdates &manager); + + void onMessage(const QByteArray &msg) /* override */; + +private: + BttvLiveUpdates &manager; +}; + +} // namespace chatterino diff --git a/src/providers/liveupdates/BasicPubSubClient.hpp b/src/providers/liveupdates/BasicPubSubClient.hpp index aa65e619..a5cbed17 100644 --- a/src/providers/liveupdates/BasicPubSubClient.hpp +++ b/src/providers/liveupdates/BasicPubSubClient.hpp @@ -1,15 +1,10 @@ #pragma once #include "common/QLogging.hpp" -#include "providers/liveupdates/BasicPubSubWebsocket.hpp" -#include "singletons/Settings.hpp" +#include "common/websockets/WebSocketPool.hpp" +#include "debug/AssertInGuiThread.hpp" #include "util/DebugCount.hpp" -#include "util/Helpers.hpp" -#include - -#include -#include #include namespace chatterino { @@ -25,51 +20,48 @@ namespace chatterino { * * @tparam Subscription see BasicPubSubManager */ -template +template class BasicPubSubClient - : public std::enable_shared_from_this> { public: + using Subscription = SubscriptionT; + // The maximum amount of subscriptions this connections can handle const size_t maxSubscriptions; - BasicPubSubClient(liveupdates::WebsocketClient &websocketClient, - liveupdates::WebsocketHandle handle, - size_t maxSubscriptions = 100) + BasicPubSubClient(size_t maxSubscriptions = 100) : maxSubscriptions(maxSubscriptions) - , websocketClient_(websocketClient) - , handle_(std::move(handle)) { } - virtual ~BasicPubSubClient() = default; - + ~BasicPubSubClient() = default; BasicPubSubClient(const BasicPubSubClient &) = delete; BasicPubSubClient(const BasicPubSubClient &&) = delete; BasicPubSubClient &operator=(const BasicPubSubClient &) = delete; BasicPubSubClient &operator=(const BasicPubSubClient &&) = delete; + /// The websocket handshake completed. + /// + /// Called from the manager in the GUI thread. + void onOpen() + { + assertInGuiThread(); + this->open_ = true; + } + + /// A message has been received. + /// + /// Called from the websocket thread. + void onMessage(const QByteArray & /*msg*/) + { + } + + void close() + { + this->ws_.close(); + } + protected: - virtual void onConnectionEstablished() - { - } - - bool send(const char *payload) - { - liveupdates::WebsocketErrorCode ec; - this->websocketClient_.send(this->handle_, payload, - websocketpp::frame::opcode::text, ec); - - if (ec) - { - qCDebug(chatterinoLiveupdates) << "Error sending message" << payload - << ":" << ec.message().c_str(); - return false; - } - - return true; - } - /** * @return true if this client subscribed to this subscription * and the current subscriptions don't exceed the maximum @@ -97,7 +89,7 @@ protected: DebugCount::increase("LiveUpdates subscriptions"); QByteArray encoded = subscription.encodeSubscribe(); - this->send(encoded); + this->ws_.sendText(encoded); return true; } @@ -117,72 +109,23 @@ protected: DebugCount::decrease("LiveUpdates subscriptions"); QByteArray encoded = subscription.encodeUnsubscribe(); - this->send(encoded); + this->ws_.sendText(encoded); return true; } - void close(const std::string &reason, - websocketpp::close::status::value code = - websocketpp::close::status::normal) + bool isOpen() const { - liveupdates::WebsocketErrorCode ec; - - auto conn = this->websocketClient_.get_con_from_hdl(this->handle_, ec); - if (ec) - { - qCDebug(chatterinoLiveupdates) - << "Error getting connection:" << ec.message().c_str(); - return; - } - - conn->close(code, reason, ec); - if (ec) - { - qCDebug(chatterinoLiveupdates) - << "Error closing:" << ec.message().c_str(); - return; - } + return this->open_; } - bool isStarted() const - { - return this->started_.load(std::memory_order_acquire); - } - - /** - * @brief Will be called when the clients has been requested to stop - * - * Derived classes can override this to implement their own shutdown behaviour - */ - virtual void stopImpl() - { - } - - liveupdates::WebsocketClient &websocketClient_; - private: - void start() - { - assert(!this->isStarted()); - this->started_.store(true, std::memory_order_release); - this->onConnectionEstablished(); - } - - void stop() - { - assert(this->isStarted()); - this->started_.store(false, std::memory_order_release); - - this->stopImpl(); - } - - liveupdates::WebsocketHandle handle_; + WebSocketHandle ws_; std::unordered_set subscriptions_; - std::atomic started_{false}; + bool open_ = false; - template + template friend class BasicPubSubManager; }; diff --git a/src/providers/liveupdates/BasicPubSubListener.hpp b/src/providers/liveupdates/BasicPubSubListener.hpp new file mode 100644 index 00000000..9d4055d6 --- /dev/null +++ b/src/providers/liveupdates/BasicPubSubListener.hpp @@ -0,0 +1,66 @@ +#pragma once + +#include "common/websockets/WebSocketPool.hpp" +#include "util/PostToThread.hpp" + +#include +#include + +#include + +namespace chatterino { + +template +struct BasicPubSubListener : public WebSocketListener { + BasicPubSubListener(std::weak_ptr client, + QPointer manager, size_t id) + : client(std::move(client)) + , manager(std::move(manager)) + , id(id) + { + } + + void onOpen() override + { + runInGuiThread([manager = this->manager, id = this->id] { + if (manager) + { + manager->onConnectionOpen(id); + } + }); + } + + void onTextMessage(QByteArray msg) override + { + auto sp = this->client.lock(); + if (sp) + { + sp->onMessage(msg); + } + } + + void onBinaryMessage(QByteArray msg) override + { + auto sp = this->client.lock(); + if (sp) + { + sp->onMessage(msg); + } + } + + void onClose(std::unique_ptr /* self */) override + { + runInGuiThread([manager = this->manager, id = this->id] { + if (manager) + { + manager->onConnectionClose(id); + } + }); + } + + std::weak_ptr client; + QPointer manager; + size_t id; +}; + +} // namespace chatterino diff --git a/src/providers/liveupdates/BasicPubSubManager.hpp b/src/providers/liveupdates/BasicPubSubManager.hpp index 7c3ca147..66c922da 100644 --- a/src/providers/liveupdates/BasicPubSubManager.hpp +++ b/src/providers/liveupdates/BasicPubSubManager.hpp @@ -1,105 +1,81 @@ #pragma once #include "common/QLogging.hpp" -#include "common/Version.hpp" -#include "providers/liveupdates/BasicPubSubClient.hpp" -#include "providers/liveupdates/BasicPubSubWebsocket.hpp" -#include "providers/NetworkConfigurationProvider.hpp" -#include "providers/twitch/PubSubHelpers.hpp" +#include "common/websockets/WebSocketPool.hpp" +#include "providers/liveupdates/BasicPubSubListener.hpp" +#include "providers/liveupdates/Diag.hpp" #include "util/DebugCount.hpp" #include "util/ExponentialBackoff.hpp" -#include "util/OnceFlag.hpp" -#include "util/RenameThread.hpp" -#include -#include -#include -#include -#include -#include +#include +#include #include -#include #include -#include -#include #include -#include #include #include #include namespace chatterino { +namespace liveupdates { + +template +concept IsManager = requires(Manager &manager) { + { manager.makeClient() } -> std::same_as>; +}; + +template +concept IsClient = requires(Client &client, const QByteArray &msg) { + { client.onOpen() } -> std::same_as; + { client.onMessage(msg) } -> std::same_as; + { client.close() } -> std::same_as; +}; + +} // namespace liveupdates + /** * This class is the basis for connecting and interacting with * simple PubSub servers over the Websocket protocol. * It acts as a pool for connections (see BasicPubSubClient). * - * You can customize the clients, by creating your custom - * client in ::createClient. - * - * You **must** implement #onMessage. The method gets called for every - * received message on every connection. - * If you want to get the connection this message was received on, - * use #findClient. + * You **must** implement a method `makeClient()` that returns a shared pointer + * of the client. * * You must expose your own subscribe and unsubscribe methods * (e.g. [un-]subscribeTopic). * This manager does not keep track of the subscriptions. * - * @tparam Subscription - * The subscription has the following requirements: - * It must have the methods QByteArray encodeSubscribe(), - * and QByteArray encodeUnsubscribe(). - * It must have an overload for - * QDebug &operator<< (see tests/src/BasicPubSub.cpp), - * a specialization for std::hash, - * and and overload for operator== and operator!=. + * @tparam Derived + * The derived class. Used to dispatch to makeClient(). + * + * @tparam ClientT + * The client type. Must confirm to the IsClient above. Use BasicPubSubClient + * for a common implementation. Used to dispatch to the correct methods there + * and to get the subscription type. * * @see BasicPubSubClient */ -template -class BasicPubSubManager +template +class BasicPubSubManager : public QObject { public: + using Subscription = ClientT::Subscription; + using Client = ClientT; + BasicPubSubManager(QString host, QString shortName) - : host_(std::move(host)) - , shortName_(std::move(shortName)) + : pool_(std::make_optional(shortName)) + , host_(std::move(host)) { - this->websocketClient_.set_access_channels( - websocketpp::log::alevel::all); - this->websocketClient_.clear_access_channels( - websocketpp::log::alevel::frame_payload | - websocketpp::log::alevel::frame_header); - - this->websocketClient_.init_asio(); - - // SSL Handshake - this->websocketClient_.set_tls_init_handler([this](auto hdl) { - return this->onTLSInit(hdl); - }); - - this->websocketClient_.set_message_handler([this](auto hdl, auto msg) { - this->onMessage(hdl, msg); - }); - this->websocketClient_.set_open_handler([this](auto hdl) { - this->onConnectionOpen(hdl); - }); - this->websocketClient_.set_close_handler([this](auto hdl) { - this->onConnectionClose(hdl); - }); - this->websocketClient_.set_fail_handler([this](auto hdl) { - this->onConnectionFail(hdl); - }); - this->websocketClient_.set_user_agent( - QStringLiteral("Chatterino/%1 (%2)") - .arg(Version::instance().version(), - Version::instance().commitHash()) - .toStdString()); + // We do this here, because `Derived` needs to be a complete type. If we + // did it as a requires clause on the class, the type would be + // incomplete. + static_assert(liveupdates::IsManager); + static_assert(liveupdates::IsClient); } - virtual ~BasicPubSubManager() + ~BasicPubSubManager() override { // The derived class must call stop in its destructor assert(this->stopping_); @@ -111,28 +87,7 @@ public: BasicPubSubManager &operator=(const BasicPubSubManager &&) = delete; /** This is only used for testing. */ - struct { - std::atomic connectionsClosed{0}; - std::atomic connectionsOpened{0}; - std::atomic connectionsFailed{0}; - } diag; - - void start() - { - this->work_ = std::make_shared>( - this->websocketClient_.get_io_service().get_executor()); - this->mainThread_.reset(new std::thread([this] { - // make sure we set in any case, even exceptions - auto guard = qScopeGuard([&] { - this->stoppedFlag_.set(); - }); - - runThread(); - })); - - renameThread(*this->mainThread_.get(), "BPSM-" % this->shortName_); - } + liveupdates::Diag diag; void stop() { @@ -142,74 +97,10 @@ public: } this->stopping_ = true; - - for (const auto &client : this->clients_) - { - client.second->close("Shutting down"); - } - - this->work_.reset(); - - if (!this->mainThread_->joinable()) - { - return; - } - - // NOTE: - // There is a case where a new client was initiated but not added to the clients list. - // We just don't join the thread & let the operating system nuke the thread if joining fails - // within 1s. - if (this->stoppedFlag_.waitFor(std::chrono::milliseconds{100})) - { - this->mainThread_->join(); - return; - } - - qCWarning(chatterinoLiveupdates) - << "Thread didn't finish within 100ms, force-stop the client"; - this->websocketClient_.stop(); - if (this->stoppedFlag_.waitFor(std::chrono::milliseconds{20})) - { - this->mainThread_->join(); - return; - } - - qCWarning(chatterinoLiveupdates) - << "Thread didn't finish after stopping"; + this->pool_.reset(); } protected: - using WebsocketMessagePtr = - websocketpp::config::asio_tls_client::message_type::ptr; - using WebsocketContextPtr = - websocketpp::lib::shared_ptr; - - virtual void onMessage(websocketpp::connection_hdl hdl, - WebsocketMessagePtr msg) = 0; - - virtual std::shared_ptr> createClient( - liveupdates::WebsocketClient &client, websocketpp::connection_hdl hdl) - { - return std::make_shared>(client, hdl); - } - - /** - * @param hdl The handle of the client. - * @return The client managing this connection, empty shared_ptr otherwise. - */ - std::shared_ptr> findClient( - websocketpp::connection_hdl hdl) - { - auto clientIt = this->clients_.find(hdl); - - if (clientIt == this->clients_.end()) - { - return {}; - } - - return clientIt->second; - } - void unsubscribe(const Subscription &subscription) { for (auto &client : this->clients_) @@ -233,8 +124,18 @@ protected: DebugCount::increase("LiveUpdates subscription backlog"); } + const std::unordered_map> &clients() const + { + return this->clients_; + } + private: - void onConnectionOpen(websocketpp::connection_hdl hdl) + Derived *derived() + { + return static_cast(this); + } + + void onConnectionOpen(size_t id) { DebugCount::increase("LiveUpdates connections"); this->addingClient_ = false; @@ -242,14 +143,8 @@ private: this->connectBackoff_.reset(); - auto client = this->createClient(this->websocketClient_, hdl); - - // We separate the starting from the constructor because we will want to use - // shared_from_this - client->start(); - - this->clients_.emplace(hdl, client); - + auto *client = resolve(id); + client->onOpen(); auto pendingSubsToTake = std::min(this->pendingSubscriptions_.size(), client->maxSubscriptions); @@ -278,92 +173,64 @@ private: } } - void onConnectionFail(websocketpp::connection_hdl hdl) + void onConnectionClose(size_t id) { - DebugCount::increase("LiveUpdates failed connections"); - this->diag.connectionsFailed.fetch_add(1, std::memory_order_acq_rel); + this->addingClient_ = false; - if (auto conn = this->websocketClient_.get_con_from_hdl(std::move(hdl))) + auto it = this->clients_.find(id); + if (it == this->clients_.end()) { - qCDebug(chatterinoLiveupdates) - << "LiveUpdates connection attempt failed (error: " - << conn->get_ec().message().c_str() << ")"; + qCWarning(chatterinoLiveupdates) << "Unknown client:" << id; + assert(false); + return; + } + + DebugCount::decrease("LiveUpdates connections"); + qCDebug(chatterinoLiveupdates) << "Connection" << id << "closed"; + + auto subs = std::move(it->second->subscriptions_); + bool wasOpen = it->second->isOpen(); + + if (wasOpen) + { + this->diag.connectionsClosed.fetch_add(1, + std::memory_order::relaxed); } else { - qCDebug(chatterinoLiveupdates) - << "LiveUpdates connection attempt failed but we can't get the " - "connection from a handle."; + this->diag.connectionsFailed.fetch_add(1, + std::memory_order::relaxed); } - this->addingClient_ = false; - if (!this->pendingSubscriptions_.empty()) + + this->clients_.erase(it); + if (this->stopping_) { - runAfter(this->websocketClient_.get_io_service(), - this->connectBackoff_.next(), [this](auto /*timer*/) { - this->addClient(); - }); + return; } - } - void onConnectionClose(websocketpp::connection_hdl hdl) - { - qCDebug(chatterinoLiveupdates) << "Connection closed"; - DebugCount::decrease("LiveUpdates connections"); - this->diag.connectionsClosed.fetch_add(1, std::memory_order_acq_rel); - - auto clientIt = this->clients_.find(hdl); - - // If this assert goes off, there's something wrong with the connection - // creation/preserving code KKona - assert(clientIt != this->clients_.end()); - - auto client = clientIt->second; - - this->clients_.erase(clientIt); - - client->stop(); - - if (!this->stopping_) + if (!wasOpen) { - for (const auto &sub : client->subscriptions_) - { - this->subscribe(sub); - } + qCWarning(chatterinoLiveupdates) + << "Retrying after" << id << "failed"; + this->pendingSubscriptions_.insert( + this->pendingSubscriptions_.end(), + std::make_move_iterator(subs.begin()), + std::make_move_iterator(subs.end())); + QTimer::singleShot(this->connectBackoff_.next(), this, [this] { + this->addClient(); + }); + return; } - } - WebsocketContextPtr onTLSInit(const websocketpp::connection_hdl & /*hdl*/) - { - WebsocketContextPtr ctx(new boost::asio::ssl::context( - boost::asio::ssl::context::tls_client)); - - try + for (const auto &sub : subs) { - ctx->set_options(boost::asio::ssl::context::default_workarounds | - boost::asio::ssl::context::no_tlsv1 | - boost::asio::ssl::context::no_tlsv1_1 | - boost::asio::ssl::context::single_dh_use); + this->subscribe(sub); } - catch (const std::exception &e) - { - qCDebug(chatterinoLiveupdates) - << "Exception caught in OnTLSInit:" << e.what(); - } - - return ctx; - } - - void runThread() - { - qCDebug(chatterinoLiveupdates) << "Start LiveUpdates manager thread"; - this->websocketClient_.run(); - qCDebug(chatterinoLiveupdates) - << "Done with LiveUpdates manager thread"; } void addClient() { - if (this->addingClient_) + if (this->addingClient_ || !this->pool_) { return; } @@ -372,20 +239,17 @@ private: this->addingClient_ = true; - websocketpp::lib::error_code ec; - auto con = this->websocketClient_.get_connection( - this->host_.toStdString(), ec); - - if (ec) - { - qCDebug(chatterinoLiveupdates) - << "Unable to establish connection:" << ec.message().c_str(); - return; - } - - NetworkConfigurationProvider::applyToWebSocket(con); - - this->websocketClient_.connect(con); + auto id = this->nextId_++; + auto client = this->derived()->makeClient(); + auto hdl = this->pool_->createSocket( + WebSocketOptions{ + .url = this->host_, + .headers = {}, + }, + std::make_unique>( + std::weak_ptr{client}, this->derived(), id)); + client->ws_ = std::move(hdl); + this->clients_.emplace(id, std::move(client)); } bool trySubscribe(const Subscription &subscription) @@ -400,29 +264,30 @@ private: return false; } + Client *resolve(size_t id) + { + auto it = this->clients_.find(id); + if (it == this->clients_.end()) + { + return nullptr; + } + return it->second.get(); + } + std::vector pendingSubscriptions_; - std::atomic addingClient_{false}; ExponentialBackoff<5> connectBackoff_{std::chrono::milliseconds(1000)}; - liveupdates::WebsocketClient websocketClient_; - std::unique_ptr mainThread_; - OnceFlag stoppedFlag_; - - std::map>, - std::owner_less> - clients_; - - std::shared_ptr> - work_{nullptr}; + std::optional pool_; + std::unordered_map> clients_; const QString host_; - /// Short name of the service (e.g. "7TV" or "BTTV") - const QString shortName_; + size_t nextId_ = 0; - bool stopping_{false}; + bool stopping_ = false; + bool addingClient_ = false; + + friend BasicPubSubListener; }; } // namespace chatterino diff --git a/src/providers/liveupdates/BasicPubSubWebsocket.hpp b/src/providers/liveupdates/BasicPubSubWebsocket.hpp deleted file mode 100644 index a22bc152..00000000 --- a/src/providers/liveupdates/BasicPubSubWebsocket.hpp +++ /dev/null @@ -1,36 +0,0 @@ -#pragma once - -#include "providers/twitch/ChatterinoWebSocketppLogger.hpp" - -#include -#include -#include -#include - -namespace chatterino { - -struct BasicPubSubConfig : public websocketpp::config::asio_tls_client { - // NOLINTBEGIN(modernize-use-using) - typedef websocketpp::log::chatterinowebsocketpplogger< - concurrency_type, websocketpp::log::elevel> - elog_type; - typedef websocketpp::log::chatterinowebsocketpplogger< - concurrency_type, websocketpp::log::alevel> - alog_type; - - struct PerMessageDeflateConfig { - }; - - typedef websocketpp::extensions::permessage_deflate::disabled< - PerMessageDeflateConfig> - permessage_deflate_type; - // NOLINTEND(modernize-use-using) -}; - -namespace liveupdates { -using WebsocketClient = websocketpp::client; -using WebsocketHandle = websocketpp::connection_hdl; -using WebsocketErrorCode = websocketpp::lib::error_code; -} // namespace liveupdates - -} // namespace chatterino diff --git a/src/providers/liveupdates/Diag.hpp b/src/providers/liveupdates/Diag.hpp new file mode 100644 index 00000000..3b7db1b3 --- /dev/null +++ b/src/providers/liveupdates/Diag.hpp @@ -0,0 +1,13 @@ +#pragma once + +#include + +namespace chatterino::liveupdates { + +struct Diag { + std::atomic connectionsClosed{0}; + std::atomic connectionsOpened{0}; + std::atomic connectionsFailed{0}; +}; + +} // namespace chatterino::liveupdates diff --git a/src/providers/seventv/SeventvEventAPI.cpp b/src/providers/seventv/SeventvEventAPI.cpp index bec8ab5f..f0ee3d4a 100644 --- a/src/providers/seventv/SeventvEventAPI.cpp +++ b/src/providers/seventv/SeventvEventAPI.cpp @@ -1,13 +1,8 @@ #include "providers/seventv/SeventvEventAPI.hpp" #include "Application.hpp" -#include "common/Literals.hpp" +#include "providers/liveupdates/BasicPubSubManager.hpp" #include "providers/seventv/eventapi/Client.hpp" -#include "providers/seventv/eventapi/Dispatch.hpp" -#include "providers/seventv/eventapi/Message.hpp" -#include "providers/seventv/SeventvBadges.hpp" -#include "providers/seventv/SeventvCosmetics.hpp" -#include "util/QMagicEnum.hpp" #include @@ -17,49 +12,118 @@ namespace chatterino { using namespace seventv; using namespace seventv::eventapi; -using namespace chatterino::literals; +using namespace Qt::StringLiterals; -SeventvEventAPI::SeventvEventAPI( - QString host, std::chrono::milliseconds defaultHeartbeatInterval) - : BasicPubSubManager(std::move(host), u"7TV"_s) - , heartbeatInterval_(defaultHeartbeatInterval) +class SeventvEventAPIPrivate + : public BasicPubSubManager { +public: + SeventvEventAPIPrivate(SeventvEventAPI &parent, QString host, + std::chrono::milliseconds defaultHeartbeatInterval); + ~SeventvEventAPIPrivate() override; + SeventvEventAPIPrivate(const SeventvEventAPIPrivate &) = delete; + SeventvEventAPIPrivate(SeventvEventAPIPrivate &&) = delete; + SeventvEventAPIPrivate &operator=(const SeventvEventAPIPrivate &) = delete; + SeventvEventAPIPrivate &operator=(SeventvEventAPIPrivate &&) = delete; + + std::shared_ptr makeClient(); + void checkHeartbeats(); + + /** emote-set ids */ + std::unordered_set subscribedEmoteSets; + /** user ids */ + std::unordered_set subscribedUsers; + /** Twitch channel ids */ + std::unordered_set subscribedTwitchChannels; + + std::chrono::milliseconds heartbeatInterval; + QTimer heartbeatTimer; + + SeventvEventAPI &parent; + + friend BasicPubSubManager; + friend SeventvEventAPI; +}; + +SeventvEventAPIPrivate::SeventvEventAPIPrivate( + SeventvEventAPI &parent, QString host, + std::chrono::milliseconds defaultHeartbeatInterval) + : BasicPubSubManager(std::move(host), u"7TV"_s) + , heartbeatInterval(defaultHeartbeatInterval) + , parent(parent) +{ + QObject::connect(&this->heartbeatTimer, &QTimer::timeout, this, + &SeventvEventAPIPrivate::checkHeartbeats); + this->heartbeatTimer.setInterval(this->heartbeatInterval); + this->heartbeatTimer.setSingleShot(false); + this->heartbeatTimer.start(); } -SeventvEventAPI::~SeventvEventAPI() +SeventvEventAPIPrivate::~SeventvEventAPIPrivate() { this->stop(); } +std::shared_ptr SeventvEventAPIPrivate::makeClient() +{ + return std::make_shared(this->parent, this->heartbeatInterval); +} + +void SeventvEventAPIPrivate::checkHeartbeats() +{ + auto minInterval = std::chrono::milliseconds::max(); + for (const auto &[id, client] : this->clients()) + { + client->checkHeartbeat(); + minInterval = std::min(minInterval, client->heartbeatInterval()); + } + if (minInterval != std::chrono::milliseconds::max()) + { + this->heartbeatInterval = minInterval; + this->heartbeatTimer.setInterval(this->heartbeatInterval); + } +} + +SeventvEventAPI::SeventvEventAPI( + QString host, std::chrono::milliseconds defaultHeartbeatInterval) + : private_(std::make_unique( + *this, std::move(host), defaultHeartbeatInterval)) +{ +} + +SeventvEventAPI::~SeventvEventAPI() = default; + void SeventvEventAPI::subscribeUser(const QString &userID, const QString &emoteSetID) { - if (!userID.isEmpty() && this->subscribedUsers_.insert(userID).second) + if (!userID.isEmpty() && + this->private_->subscribedUsers.insert(userID).second) { - this->subscribe( + this->private_->subscribe( {ObjectIDCondition{userID}, SubscriptionType::UpdateUser}); } if (!emoteSetID.isEmpty() && - this->subscribedEmoteSets_.insert(emoteSetID).second) + this->private_->subscribedEmoteSets.insert(emoteSetID).second) { - this->subscribe( + this->private_->subscribe( {ObjectIDCondition{emoteSetID}, SubscriptionType::UpdateEmoteSet}); } } void SeventvEventAPI::subscribeTwitchChannel(const QString &id) { - if (this->subscribedTwitchChannels_.insert(id).second) + if (this->private_->subscribedTwitchChannels.insert(id).second) { - this->subscribe({ + this->private_->subscribe({ ChannelCondition{id}, SubscriptionType::CreateCosmetic, }); - this->subscribe({ + this->private_->subscribe({ ChannelCondition{id}, SubscriptionType::CreateEntitlement, }); - this->subscribe({ + this->private_->subscribe({ ChannelCondition{id}, SubscriptionType::DeleteEntitlement, }); @@ -68,362 +132,49 @@ void SeventvEventAPI::subscribeTwitchChannel(const QString &id) void SeventvEventAPI::unsubscribeEmoteSet(const QString &id) { - if (this->subscribedEmoteSets_.erase(id) > 0) + if (this->private_->subscribedEmoteSets.erase(id) > 0) { - this->unsubscribe( + this->private_->unsubscribe( {ObjectIDCondition{id}, SubscriptionType::UpdateEmoteSet}); } } void SeventvEventAPI::unsubscribeUser(const QString &id) { - if (this->subscribedUsers_.erase(id) > 0) + if (this->private_->subscribedUsers.erase(id) > 0) { - this->unsubscribe( + this->private_->unsubscribe( {ObjectIDCondition{id}, SubscriptionType::UpdateUser}); } } void SeventvEventAPI::unsubscribeTwitchChannel(const QString &id) { - if (this->subscribedTwitchChannels_.erase(id) > 0) + if (this->private_->subscribedTwitchChannels.erase(id) > 0) { - this->unsubscribe({ + this->private_->unsubscribe({ ChannelCondition{id}, SubscriptionType::CreateCosmetic, }); - this->unsubscribe({ + this->private_->unsubscribe({ ChannelCondition{id}, SubscriptionType::CreateEntitlement, }); - this->unsubscribe({ + this->private_->unsubscribe({ ChannelCondition{id}, SubscriptionType::DeleteEntitlement, }); } } -std::shared_ptr> SeventvEventAPI::createClient( - liveupdates::WebsocketClient &client, websocketpp::connection_hdl hdl) +void SeventvEventAPI::stop() { - auto shared = - std::make_shared(client, hdl, this->heartbeatInterval_); - return std::static_pointer_cast>( - std::move(shared)); + this->private_->stop(); } -void SeventvEventAPI::onMessage( - websocketpp::connection_hdl hdl, - BasicPubSubManager::WebsocketMessagePtr msg) +const liveupdates::Diag &SeventvEventAPI::diag() const { - const auto &payload = QString::fromStdString(msg->get_payload()); - - auto pMessage = parseBaseMessage(payload); - - if (!pMessage) - { - qCDebug(chatterinoSeventvEventAPI) - << "Unable to parse incoming event-api message: " << payload; - return; - } - auto message = *pMessage; - switch (message.op) - { - case Opcode::Hello: { - if (auto client = this->findClient(hdl)) - { - if (auto *stvClient = dynamic_cast(client.get())) - { - stvClient->setHeartbeatInterval( - message.data["heartbeat_interval"].toInt()); - } - } - } - break; - case Opcode::Heartbeat: { - if (auto client = this->findClient(hdl)) - { - if (auto *stvClient = dynamic_cast(client.get())) - { - stvClient->handleHeartbeat(); - } - } - } - break; - case Opcode::Dispatch: { - auto dispatch = message.toInner(); - if (!dispatch) - { - qCDebug(chatterinoSeventvEventAPI) - << "Malformed dispatch" << payload; - return; - } - this->handleDispatch(*dispatch); - } - break; - case Opcode::Reconnect: { - if (auto client = this->findClient(hdl)) - { - if (auto *stvClient = dynamic_cast(client.get())) - { - stvClient->close("Reconnecting"); - } - } - } - break; - case Opcode::Ack: { - // unhandled - } - break; - default: { - qCDebug(chatterinoSeventvEventAPI) << "Unhandled op:" << payload; - } - break; - } + return this->private_->diag; } -void SeventvEventAPI::handleDispatch(const Dispatch &dispatch) -{ - switch (dispatch.type) - { - case SubscriptionType::UpdateEmoteSet: { - this->onEmoteSetUpdate(dispatch); - } - break; - case SubscriptionType::UpdateUser: { - this->onUserUpdate(dispatch); - } - break; - case SubscriptionType::CreateCosmetic: { - const CosmeticCreateDispatch cosmetic(dispatch); - if (cosmetic.validate()) - { - this->onCosmeticCreate(cosmetic); - } - else - { - qCDebug(chatterinoSeventvEventAPI) - << "Invalid cosmetic dispatch" << dispatch.body; - } - } - break; - case SubscriptionType::CreateEntitlement: { - const EntitlementCreateDeleteDispatch entitlement(dispatch); - if (entitlement.validate()) - { - this->onEntitlementCreate(entitlement); - } - else - { - qCDebug(chatterinoSeventvEventAPI) - << "Invalid entitlement create dispatch" << dispatch.body; - } - } - break; - case SubscriptionType::DeleteEntitlement: { - const EntitlementCreateDeleteDispatch entitlement(dispatch); - if (entitlement.validate()) - { - this->onEntitlementDelete(entitlement); - } - else - { - qCDebug(chatterinoSeventvEventAPI) - << "Invalid entitlement delete dispatch" << dispatch.body; - } - } - break; - case SubscriptionType::ResetEntitlement: { - // unhandled (not clear what we'd do here yet) - } - break; - case SubscriptionType::CreateEmoteSet: { - // unhandled (c2 does not support custom emote sets) - } - break; - default: { - qCDebug(chatterinoSeventvEventAPI) - << "Unknown subscription type:" - << qmagicenum::enumName(dispatch.type) - << "body:" << dispatch.body; - } - break; - } -} - -void SeventvEventAPI::onEmoteSetUpdate(const Dispatch &dispatch) -{ - // dispatchBody: { - // pushed: Array<{ key, value }>, - // pulled: Array<{ key, old_value }>, - // updated: Array<{ key, value, old_value }>, - // } - for (const auto pushedRef : dispatch.body["pushed"].toArray()) - { - auto pushed = pushedRef.toObject(); - if (pushed["key"].toString() != "emotes") - { - continue; - } - - const EmoteAddDispatch added(dispatch, pushed["value"].toObject()); - - if (added.validate()) - { - this->signals_.emoteAdded.invoke(added); - } - else - { - qCDebug(chatterinoSeventvEventAPI) - << "Invalid dispatch" << dispatch.body; - } - } - for (const auto updatedRef : dispatch.body["updated"].toArray()) - { - auto updated = updatedRef.toObject(); - if (updated["key"].toString() != "emotes") - { - continue; - } - - const EmoteUpdateDispatch update(dispatch, - updated["old_value"].toObject(), - updated["value"].toObject()); - - if (update.validate()) - { - this->signals_.emoteUpdated.invoke(update); - } - else - { - qCDebug(chatterinoSeventvEventAPI) - << "Invalid dispatch" << dispatch.body; - } - } - for (const auto pulledRef : dispatch.body["pulled"].toArray()) - { - auto pulled = pulledRef.toObject(); - if (pulled["key"].toString() != "emotes") - { - continue; - } - - const EmoteRemoveDispatch removed(dispatch, - pulled["old_value"].toObject()); - - if (removed.validate()) - { - this->signals_.emoteRemoved.invoke(removed); - } - else - { - qCDebug(chatterinoSeventvEventAPI) - << "Invalid dispatch" << dispatch.body; - } - } -} - -void SeventvEventAPI::onUserUpdate(const Dispatch &dispatch) -{ - // dispatchBody: { - // updated: Array<{ key, value: Array<{key, value}> }> - // } - for (const auto updatedRef : dispatch.body["updated"].toArray()) - { - auto updated = updatedRef.toObject(); - if (updated["key"].toString() != "connections") - { - continue; - } - for (const auto valueRef : updated["value"].toArray()) - { - auto value = valueRef.toObject(); - if (value["key"].toString() != "emote_set") - { - continue; - } - - const UserConnectionUpdateDispatch update( - dispatch, value, (size_t)updated["index"].toInt()); - - if (update.validate()) - { - this->signals_.userUpdated.invoke(update); - } - else - { - qCDebug(chatterinoSeventvEventAPI) - << "Invalid dispatch" << dispatch.body; - } - } - } -} - -// NOLINTBEGIN(readability-convert-member-functions-to-static) - -void SeventvEventAPI::onCosmeticCreate(const CosmeticCreateDispatch &cosmetic) -{ - auto *app = tryGetApp(); - if (!app) - { - return; // shutting down - } - - auto *badges = app->getSeventvBadges(); - switch (cosmetic.kind) - { - case CosmeticKind::Badge: { - badges->registerBadge(cosmetic.data); - } - break; - default: - break; - } -} - -void SeventvEventAPI::onEntitlementCreate( - const EntitlementCreateDeleteDispatch &entitlement) -{ - auto *app = tryGetApp(); - if (!app) - { - return; // shutting down - } - - auto *badges = app->getSeventvBadges(); - switch (entitlement.kind) - { - case CosmeticKind::Badge: { - badges->assignBadgeToUser(entitlement.refID, - UserId{entitlement.userID}); - } - break; - default: - break; - } -} - -void SeventvEventAPI::onEntitlementDelete( - const EntitlementCreateDeleteDispatch &entitlement) -{ - auto *app = tryGetApp(); - if (!app) - { - return; // shutting down - } - - auto *badges = app->getSeventvBadges(); - switch (entitlement.kind) - { - case CosmeticKind::Badge: { - badges->clearBadgeFromUser(entitlement.refID, - UserId{entitlement.userID}); - } - break; - default: - break; - } -} -// NOLINTEND(readability-convert-member-functions-to-static) - } // namespace chatterino diff --git a/src/providers/seventv/SeventvEventAPI.hpp b/src/providers/seventv/SeventvEventAPI.hpp index ea6159bf..36e9b913 100644 --- a/src/providers/seventv/SeventvEventAPI.hpp +++ b/src/providers/seventv/SeventvEventAPI.hpp @@ -1,39 +1,36 @@ #pragma once -#include "providers/liveupdates/BasicPubSubClient.hpp" -#include "providers/liveupdates/BasicPubSubManager.hpp" -#include "providers/seventv/eventapi/Subscription.hpp" -#include "util/QStringHash.hpp" - #include +#include + +#include namespace chatterino { +namespace liveupdates { +struct Diag; +} // namespace liveupdates + namespace seventv::eventapi { -struct Dispatch; struct EmoteAddDispatch; struct EmoteUpdateDispatch; struct EmoteRemoveDispatch; struct UserConnectionUpdateDispatch; -struct CosmeticCreateDispatch; -struct EntitlementCreateDeleteDispatch; } // namespace seventv::eventapi class SeventvBadges; +class SeventvEventAPIPrivate; class SeventvEventAPI - : public BasicPubSubManager { template - using Signal = - pajlada::Signals::Signal; // type-id is vector> + using Signal = pajlada::Signals::Signal; public: SeventvEventAPI(QString host, std::chrono::milliseconds defaultHeartbeatInterval = std::chrono::milliseconds(25000)); - - ~SeventvEventAPI() override; + ~SeventvEventAPI(); struct { Signal emoteAdded; @@ -65,34 +62,19 @@ public: /** Unsubscribes from cosmetics and entitlements in a Twitch channel */ void unsubscribeTwitchChannel(const QString &id); -protected: - std::shared_ptr> - createClient(liveupdates::WebsocketClient &client, - websocketpp::connection_hdl hdl) override; - void onMessage( - websocketpp::connection_hdl hdl, - BasicPubSubManager::WebsocketMessagePtr - msg) override; + /// Stop the manager + /// + /// Used in tests to check that connections are closed (through #diag()). + /// Otherwise, calling the destructor is sufficient. + void stop(); + + /// Statistics about the opened/closed connections and received messages + /// + /// Used in tests. + const liveupdates::Diag &diag() const; private: - void handleDispatch(const seventv::eventapi::Dispatch &dispatch); - - void onEmoteSetUpdate(const seventv::eventapi::Dispatch &dispatch); - void onUserUpdate(const seventv::eventapi::Dispatch &dispatch); - void onCosmeticCreate( - const seventv::eventapi::CosmeticCreateDispatch &cosmetic); - void onEntitlementCreate( - const seventv::eventapi::EntitlementCreateDeleteDispatch &entitlement); - void onEntitlementDelete( - const seventv::eventapi::EntitlementCreateDeleteDispatch &entitlement); - - /** emote-set ids */ - std::unordered_set subscribedEmoteSets_; - /** user ids */ - std::unordered_set subscribedUsers_; - /** Twitch channel ids */ - std::unordered_set subscribedTwitchChannels_; - std::chrono::milliseconds heartbeatInterval_; + std::unique_ptr private_; }; } // namespace chatterino diff --git a/src/providers/seventv/eventapi/Client.cpp b/src/providers/seventv/eventapi/Client.cpp index 84533a06..6382b35c 100644 --- a/src/providers/seventv/eventapi/Client.cpp +++ b/src/providers/seventv/eventapi/Client.cpp @@ -1,73 +1,343 @@ #include "providers/seventv/eventapi/Client.hpp" +#include "Application.hpp" +#include "providers/seventv/eventapi/Dispatch.hpp" +#include "providers/seventv/eventapi/Message.hpp" #include "providers/seventv/eventapi/Subscription.hpp" -#include "providers/twitch/PubSubHelpers.hpp" +#include "providers/seventv/SeventvBadges.hpp" +#include "providers/seventv/SeventvEventAPI.hpp" +#include "util/QMagicEnum.hpp" -#include +#include namespace chatterino::seventv::eventapi { -Client::Client(liveupdates::WebsocketClient &websocketClient, - liveupdates::WebsocketHandle handle, +Client::Client(SeventvEventAPI &manager, std::chrono::milliseconds heartbeatInterval) - : BasicPubSubClient(websocketClient, std::move(handle)) + : BasicPubSubClient(100) , lastHeartbeat_(std::chrono::steady_clock::now()) , heartbeatInterval_(heartbeatInterval) - , heartbeatTimer_(std::make_shared( - this->websocketClient_.get_io_service())) + , manager_(manager) { } -void Client::stopImpl() -{ - this->heartbeatTimer_->cancel(); -} - -void Client::onConnectionEstablished() +void Client::onOpen() { + BasicPubSubClient::onOpen(); this->lastHeartbeat_.store(std::chrono::steady_clock::now(), - std::memory_order_release); - this->checkHeartbeat(); + std::memory_order::relaxed); } -void Client::setHeartbeatInterval(int intervalMs) +void Client::onMessage(const QByteArray &msg) { - qCDebug(chatterinoSeventvEventAPI) - << "Setting expected heartbeat interval to" << intervalMs << "ms"; - this->heartbeatInterval_ = std::chrono::milliseconds(intervalMs); -} + auto pMessage = parseBaseMessage(msg); -void Client::handleHeartbeat() -{ - this->lastHeartbeat_.store(std::chrono::steady_clock::now(), - std::memory_order_release); + if (!pMessage) + { + qCDebug(chatterinoSeventvEventAPI) + << "Unable to parse incoming event-api message: " << msg; + return; + } + auto message = *pMessage; + switch (message.op) + { + case Opcode::Hello: { + this->heartbeatInterval_.store( + std::chrono::milliseconds{ + message.data["heartbeat_interval"].toInt()}, + std::memory_order::relaxed); + } + break; + case Opcode::Heartbeat: { + this->lastHeartbeat_.store(std::chrono::steady_clock::now(), + std::memory_order::relaxed); + } + break; + case Opcode::Dispatch: { + auto dispatch = message.toInner(); + if (!dispatch) + { + qCDebug(chatterinoSeventvEventAPI) + << "Malformed dispatch" << msg; + return; + } + this->handleDispatch(*dispatch); + } + break; + case Opcode::Reconnect: { + this->close(); + } + break; + case Opcode::Ack: { + // unhandled + } + break; + default: { + qCDebug(chatterinoSeventvEventAPI) << "Unhandled op:" << msg; + } + break; + } } void Client::checkHeartbeat() { - // Following the heartbeat docs, a connection is dead - // after three missed heartbeats. - // https://github.com/SevenTV/EventAPI/tree/ca4ff15cc42b89560fa661a76c5849047763d334#heartbeat - assert(this->isStarted()); - if ((std::chrono::steady_clock::now() - this->lastHeartbeat_.load()) > - 3 * this->heartbeatInterval_) + if ((std::chrono::steady_clock::now() - + this->lastHeartbeat_.load(std::memory_order::relaxed)) > + this->heartbeatInterval() * 3) { - qCDebug(chatterinoSeventvEventAPI) - << "Didn't receive a heartbeat in time, disconnecting!"; - this->close("Didn't receive a heartbeat in time"); - - return; + qCDebug(chatterinoSeventvEventAPI) << "Hearbeat timed out"; + this->close(); } - - auto self = std::dynamic_pointer_cast(this->shared_from_this()); - - runAfter(this->heartbeatTimer_, this->heartbeatInterval_, [self](auto) { - if (!self->isStarted()) - { - return; - } - self->checkHeartbeat(); - }); } +std::chrono::milliseconds Client::heartbeatInterval() const +{ + return this->heartbeatInterval_.load(std::memory_order::relaxed); +} + +void Client::handleDispatch(const Dispatch &dispatch) +{ + switch (dispatch.type) + { + case SubscriptionType::UpdateEmoteSet: { + this->onEmoteSetUpdate(dispatch); + } + break; + case SubscriptionType::UpdateUser: { + this->onUserUpdate(dispatch); + } + break; + case SubscriptionType::CreateCosmetic: { + const CosmeticCreateDispatch cosmetic(dispatch); + if (cosmetic.validate()) + { + this->onCosmeticCreate(cosmetic); + } + else + { + qCDebug(chatterinoSeventvEventAPI) + << "Invalid cosmetic dispatch" << dispatch.body; + } + } + break; + case SubscriptionType::CreateEntitlement: { + const EntitlementCreateDeleteDispatch entitlement(dispatch); + if (entitlement.validate()) + { + this->onEntitlementCreate(entitlement); + } + else + { + qCDebug(chatterinoSeventvEventAPI) + << "Invalid entitlement create dispatch" << dispatch.body; + } + } + break; + case SubscriptionType::DeleteEntitlement: { + const EntitlementCreateDeleteDispatch entitlement(dispatch); + if (entitlement.validate()) + { + this->onEntitlementDelete(entitlement); + } + else + { + qCDebug(chatterinoSeventvEventAPI) + << "Invalid entitlement delete dispatch" << dispatch.body; + } + } + break; + // NOLINTNEXTLINE(bugprone-branch-clone) + case SubscriptionType::ResetEntitlement: { + // unhandled (not clear what we'd do here yet) + } + break; + case SubscriptionType::CreateEmoteSet: { + // unhandled (c2 does not support custom emote sets) + } + break; + default: { + qCDebug(chatterinoSeventvEventAPI) + << "Unknown subscription type:" + << qmagicenum::enumName(dispatch.type) + << "body:" << dispatch.body; + } + break; + } +} + +void Client::onEmoteSetUpdate(const Dispatch &dispatch) +{ + // dispatchBody: { + // pushed: Array<{ key, value }>, + // pulled: Array<{ key, old_value }>, + // updated: Array<{ key, value, old_value }>, + // } + for (const auto pushedRef : dispatch.body["pushed"].toArray()) + { + auto pushed = pushedRef.toObject(); + if (pushed["key"].toString() != "emotes") + { + continue; + } + + const EmoteAddDispatch added(dispatch, pushed["value"].toObject()); + + if (added.validate()) + { + this->manager_.signals_.emoteAdded.invoke(added); + } + else + { + qCDebug(chatterinoSeventvEventAPI) + << "Invalid dispatch" << dispatch.body; + } + } + for (const auto updatedRef : dispatch.body["updated"].toArray()) + { + auto updated = updatedRef.toObject(); + if (updated["key"].toString() != "emotes") + { + continue; + } + + const EmoteUpdateDispatch update(dispatch, + updated["old_value"].toObject(), + updated["value"].toObject()); + + if (update.validate()) + { + this->manager_.signals_.emoteUpdated.invoke(update); + } + else + { + qCDebug(chatterinoSeventvEventAPI) + << "Invalid dispatch" << dispatch.body; + } + } + for (const auto pulledRef : dispatch.body["pulled"].toArray()) + { + auto pulled = pulledRef.toObject(); + if (pulled["key"].toString() != "emotes") + { + continue; + } + + const EmoteRemoveDispatch removed(dispatch, + pulled["old_value"].toObject()); + + if (removed.validate()) + { + this->manager_.signals_.emoteRemoved.invoke(removed); + } + else + { + qCDebug(chatterinoSeventvEventAPI) + << "Invalid dispatch" << dispatch.body; + } + } +} + +void Client::onUserUpdate(const Dispatch &dispatch) +{ + // dispatchBody: { + // updated: Array<{ key, value: Array<{key, value}> }> + // } + for (const auto updatedRef : dispatch.body["updated"].toArray()) + { + auto updated = updatedRef.toObject(); + if (updated["key"].toString() != "connections") + { + continue; + } + for (const auto valueRef : updated["value"].toArray()) + { + auto value = valueRef.toObject(); + if (value["key"].toString() != "emote_set") + { + continue; + } + + const UserConnectionUpdateDispatch update( + dispatch, value, (size_t)updated["index"].toInt()); + + if (update.validate()) + { + this->manager_.signals_.userUpdated.invoke(update); + } + else + { + qCDebug(chatterinoSeventvEventAPI) + << "Invalid dispatch" << dispatch.body; + } + } + } +} + +// NOLINTBEGIN(readability-convert-member-functions-to-static) + +void Client::onCosmeticCreate(const CosmeticCreateDispatch &cosmetic) +{ + auto *app = tryGetApp(); + if (!app) + { + return; // shutting down + } + + auto *badges = app->getSeventvBadges(); + switch (cosmetic.kind) + { + case CosmeticKind::Badge: { + badges->registerBadge(cosmetic.data); + } + break; + default: + break; + } +} + +void Client::onEntitlementCreate( + const EntitlementCreateDeleteDispatch &entitlement) +{ + auto *app = tryGetApp(); + if (!app) + { + return; // shutting down + } + + auto *badges = app->getSeventvBadges(); + switch (entitlement.kind) + { + case CosmeticKind::Badge: { + badges->assignBadgeToUser(entitlement.refID, + UserId{entitlement.userID}); + } + break; + default: + break; + } +} + +void Client::onEntitlementDelete( + const EntitlementCreateDeleteDispatch &entitlement) +{ + auto *app = tryGetApp(); + if (!app) + { + return; // shutting down + } + + auto *badges = app->getSeventvBadges(); + switch (entitlement.kind) + { + case CosmeticKind::Badge: { + badges->clearBadgeFromUser(entitlement.refID, + UserId{entitlement.userID}); + } + break; + default: + break; + } +} +// NOLINTEND(readability-convert-member-functions-to-static) + } // namespace chatterino::seventv::eventapi diff --git a/src/providers/seventv/eventapi/Client.hpp b/src/providers/seventv/eventapi/Client.hpp index ecbc7cb7..dad66758 100644 --- a/src/providers/seventv/eventapi/Client.hpp +++ b/src/providers/seventv/eventapi/Client.hpp @@ -5,6 +5,8 @@ // of std::hash for Subscription #include "providers/seventv/eventapi/Subscription.hpp" +#include + namespace chatterino { class SeventvEventAPI; @@ -12,31 +14,38 @@ class SeventvEventAPI; namespace chatterino::seventv::eventapi { -class Client : public BasicPubSubClient +struct Dispatch; +struct CosmeticCreateDispatch; +struct EntitlementCreateDeleteDispatch; + +class Client : public BasicPubSubClient, + std::enable_shared_from_this { public: - Client(liveupdates::WebsocketClient &websocketClient, - liveupdates::WebsocketHandle handle, + Client(SeventvEventAPI &manager, std::chrono::milliseconds heartbeatInterval); - void stopImpl() override; + void onOpen() /* override */; + void onMessage(const QByteArray &msg) /* override */; - void setHeartbeatInterval(int intervalMs); - void handleHeartbeat(); - -protected: - void onConnectionEstablished() override; + std::chrono::milliseconds heartbeatInterval() const; + void checkHeartbeat(); private: - void checkHeartbeat(); + void handleDispatch(const Dispatch &dispatch); + + void onEmoteSetUpdate(const Dispatch &dispatch); + void onUserUpdate(const Dispatch &dispatch); + void onCosmeticCreate(const CosmeticCreateDispatch &cosmetic); + void onEntitlementCreate( + const EntitlementCreateDeleteDispatch &entitlement); + void onEntitlementDelete( + const EntitlementCreateDeleteDispatch &entitlement); std::atomic> lastHeartbeat_; - // This will be set once on the welcome message. - std::chrono::milliseconds heartbeatInterval_; - std::shared_ptr heartbeatTimer_; - - friend SeventvEventAPI; + std::atomic heartbeatInterval_; + SeventvEventAPI &manager_; }; } // namespace chatterino::seventv::eventapi diff --git a/src/providers/seventv/eventapi/Message.cpp b/src/providers/seventv/eventapi/Message.cpp index f3b59f7a..29930616 100644 --- a/src/providers/seventv/eventapi/Message.cpp +++ b/src/providers/seventv/eventapi/Message.cpp @@ -8,9 +8,9 @@ Message::Message(QJsonObject _json) { } -std::optional parseBaseMessage(const QString &blob) +std::optional parseBaseMessage(const QByteArray &blob) { - QJsonDocument jsonDoc(QJsonDocument::fromJson(blob.toUtf8())); + QJsonDocument jsonDoc(QJsonDocument::fromJson(blob)); if (jsonDoc.isNull()) { diff --git a/src/providers/seventv/eventapi/Message.hpp b/src/providers/seventv/eventapi/Message.hpp index 5a3eebc9..d32f1ea8 100644 --- a/src/providers/seventv/eventapi/Message.hpp +++ b/src/providers/seventv/eventapi/Message.hpp @@ -28,6 +28,6 @@ std::optional Message::toInner() return InnerClass{this->data}; } -std::optional parseBaseMessage(const QString &blob); +std::optional parseBaseMessage(const QByteArray &blob); } // namespace chatterino::seventv::eventapi diff --git a/src/providers/twitch/TwitchIrcServer.cpp b/src/providers/twitch/TwitchIrcServer.cpp index 3cde0663..86293b0a 100644 --- a/src/providers/twitch/TwitchIrcServer.cpp +++ b/src/providers/twitch/TwitchIrcServer.cpp @@ -12,6 +12,7 @@ #include "messages/MessageBuilder.hpp" #include "providers/bttv/BttvEmotes.hpp" #include "providers/bttv/BttvLiveUpdates.hpp" +#include "providers/bttv/liveupdates/BttvLiveUpdateMessages.hpp" // IWYU pragma: keep #include "providers/ffz/FfzEmotes.hpp" #include "providers/irc/IrcConnection2.hpp" #include "providers/seventv/eventapi/Dispatch.hpp" // IWYU pragma: keep @@ -880,8 +881,6 @@ void TwitchIrcServer::initEventAPIs(BttvLiveUpdates *bttvLiveUpdates, }, this); }); - - bttvLiveUpdates->start(); } else { @@ -932,8 +931,6 @@ void TwitchIrcServer::initEventAPIs(BttvLiveUpdates *bttvLiveUpdates, chan.updateSeventvUser(data); }); }); - - seventvEventAPI->start(); } else { diff --git a/tests/src/BasicPubSub.cpp b/tests/src/BasicPubSub.cpp index 3b5c6b8f..e6de95f7 100644 --- a/tests/src/BasicPubSub.cpp +++ b/tests/src/BasicPubSub.cpp @@ -1,3 +1,4 @@ +#include "mocks/BaseApplication.hpp" #include "providers/liveupdates/BasicPubSubClient.hpp" #include "providers/liveupdates/BasicPubSubManager.hpp" #include "Test.hpp" @@ -6,6 +7,7 @@ #include #include #include +#include #include #include @@ -14,6 +16,8 @@ using namespace chatterino; using namespace std::chrono_literals; +namespace { + struct DummySubscription { int type; QString condition; @@ -54,6 +58,8 @@ struct DummySubscription { } }; +} // namespace + namespace std { template <> struct hash { @@ -64,7 +70,24 @@ struct hash { }; } // namespace std -class MyManager : public BasicPubSubManager +namespace { + +class MyManager; +class MyClient : public BasicPubSubClient +{ +public: + MyClient(MyManager &manager) + : manager(manager) + { + } + + void onMessage(const QByteArray &msg) /* override */; + +private: + MyManager &manager; +}; + +class MyManager : public BasicPubSubManager { public: MyManager(QString host) @@ -97,33 +120,34 @@ public: this->unsubscribe(sub); } -protected: - void onMessage( - websocketpp::connection_hdl /*hdl*/, - BasicPubSubManager::WebsocketMessagePtr msg) override + std::shared_ptr makeClient() { - std::lock_guard guard(this->messageMtx_); - this->messagesReceived.fetch_add(1, std::memory_order_acq_rel); - this->messageQueue_.emplace_back( - QString::fromStdString(msg->get_payload())); + return std::make_shared(*this); } private: std::mutex messageMtx_; std::deque messageQueue_; + + friend MyClient; }; +void MyClient::onMessage(const QByteArray &msg) +{ + std::lock_guard guard(this->manager.messageMtx_); + this->manager.messagesReceived.fetch_add(1, std::memory_order_acq_rel); + this->manager.messageQueue_.emplace_back(QString::fromUtf8(msg)); +} + +} // namespace + TEST(BasicPubSub, SubscriptionCycle) { + mock::BaseApplication app; const QString host("wss://127.0.0.1:9050/liveupdates/sub-unsub"); MyManager manager(host); - manager.start(); - - std::this_thread::sleep_for(50ms); manager.sub({1, "foo"}); - std::this_thread::sleep_for(500ms); - - ASSERT_EQ(manager.diag.connectionsOpened, 1); + QTest::qWait(500); ASSERT_EQ(manager.diag.connectionsClosed, 0); ASSERT_EQ(manager.diag.connectionsFailed, 0); ASSERT_EQ(manager.messagesReceived, 1); @@ -131,7 +155,7 @@ TEST(BasicPubSub, SubscriptionCycle) ASSERT_EQ(manager.popMessage(), QString("ack-sub-1-foo")); manager.unsub({1, "foo"}); - std::this_thread::sleep_for(50ms); + QTest::qWait(50); ASSERT_EQ(manager.diag.connectionsOpened, 1); ASSERT_EQ(manager.diag.connectionsClosed, 0); @@ -140,6 +164,9 @@ TEST(BasicPubSub, SubscriptionCycle) ASSERT_EQ(manager.popMessage(), QString("ack-unsub-1-foo")); manager.stop(); + // after exactly one event loop iteration, we should see updated counters + QCoreApplication::processEvents(QEventLoop::AllEvents); + QCoreApplication::sendPostedEvents(nullptr, QEvent::DeferredDelete); ASSERT_EQ(manager.diag.connectionsOpened, 1); ASSERT_EQ(manager.diag.connectionsClosed, 1); diff --git a/tests/src/BttvLiveUpdates.cpp b/tests/src/BttvLiveUpdates.cpp index 2d238f9b..3bebd026 100644 --- a/tests/src/BttvLiveUpdates.cpp +++ b/tests/src/BttvLiveUpdates.cpp @@ -1,8 +1,12 @@ #include "providers/bttv/BttvLiveUpdates.hpp" +#include "mocks/BaseApplication.hpp" +#include "providers/bttv/liveupdates/BttvLiveUpdateMessages.hpp" +#include "providers/liveupdates/Diag.hpp" #include "Test.hpp" #include +#include #include #include @@ -15,9 +19,10 @@ const QString TARGET_USER_NAME = "Alien"; TEST(BttvLiveUpdates, AllEvents) { + mock::BaseApplication app; + const QString host("wss://127.0.0.1:9050/liveupdates/bttv/all-events"); chatterino::BttvLiveUpdates liveUpdates(host); - liveUpdates.start(); std::optional addMessage; std::optional updateMessage; @@ -33,13 +38,12 @@ TEST(BttvLiveUpdates, AllEvents) removeMessage = m; }); - std::this_thread::sleep_for(50ms); liveUpdates.joinChannel(TARGET_USER_ID, TARGET_USER_NAME); - std::this_thread::sleep_for(500ms); + QTest::qWait(500); - ASSERT_EQ(liveUpdates.diag.connectionsOpened, 1); - ASSERT_EQ(liveUpdates.diag.connectionsClosed, 0); - ASSERT_EQ(liveUpdates.diag.connectionsFailed, 0); + ASSERT_EQ(liveUpdates.diag().connectionsOpened, 1); + ASSERT_EQ(liveUpdates.diag().connectionsClosed, 0); + ASSERT_EQ(liveUpdates.diag().connectionsFailed, 0); auto add = *addMessage; ASSERT_EQ(add.channelID, TARGET_USER_ID); @@ -56,7 +60,11 @@ TEST(BttvLiveUpdates, AllEvents) ASSERT_EQ(rem.emoteID, QString("55898e122612142e6aaa935b")); liveUpdates.stop(); - ASSERT_EQ(liveUpdates.diag.connectionsOpened, 1); - ASSERT_EQ(liveUpdates.diag.connectionsClosed, 1); - ASSERT_EQ(liveUpdates.diag.connectionsFailed, 0); + // after exactly one event loop iteration, we should see updated counters + QCoreApplication::processEvents(QEventLoop::AllEvents); + QCoreApplication::sendPostedEvents(nullptr, QEvent::DeferredDelete); + + ASSERT_EQ(liveUpdates.diag().connectionsOpened, 1); + ASSERT_EQ(liveUpdates.diag().connectionsClosed, 1); + ASSERT_EQ(liveUpdates.diag().connectionsFailed, 0); } diff --git a/tests/src/IrcMessageHandler.cpp b/tests/src/IrcMessageHandler.cpp index c6d9fa99..9e56c045 100644 --- a/tests/src/IrcMessageHandler.cpp +++ b/tests/src/IrcMessageHandler.cpp @@ -37,6 +37,7 @@ #include #include #include +#include #include #include diff --git a/tests/src/SeventvEventAPI.cpp b/tests/src/SeventvEventAPI.cpp index 2c3e2b3a..7a781369 100644 --- a/tests/src/SeventvEventAPI.cpp +++ b/tests/src/SeventvEventAPI.cpp @@ -1,11 +1,14 @@ #include "providers/seventv/SeventvEventAPI.hpp" +#include "mocks/BaseApplication.hpp" +#include "providers/liveupdates/Diag.hpp" #include "providers/seventv/eventapi/Client.hpp" #include "providers/seventv/eventapi/Dispatch.hpp" #include "providers/seventv/eventapi/Message.hpp" #include "Test.hpp" #include +#include #include @@ -19,9 +22,9 @@ const QString TARGET_USER_ID = "60b39e943e203cc169dfc106"; TEST(SeventvEventAPI, AllEvents) { + mock::BaseApplication app; const QString host("wss://127.0.0.1:9050/liveupdates/seventv/all-events"); SeventvEventAPI eventAPI(host, std::chrono::milliseconds(1000)); - eventAPI.start(); std::optional addDispatch; std::optional updateDispatch; @@ -41,13 +44,12 @@ TEST(SeventvEventAPI, AllEvents) userDispatch = d; }); - std::this_thread::sleep_for(50ms); eventAPI.subscribeUser("", EMOTE_SET_A); - std::this_thread::sleep_for(500ms); + QTest::qWait(500); - ASSERT_EQ(eventAPI.diag.connectionsOpened, 1); - ASSERT_EQ(eventAPI.diag.connectionsClosed, 0); - ASSERT_EQ(eventAPI.diag.connectionsFailed, 0); + ASSERT_EQ(eventAPI.diag().connectionsOpened, 1); + ASSERT_EQ(eventAPI.diag().connectionsClosed, 0); + ASSERT_EQ(eventAPI.diag().connectionsFailed, 0); auto add = *addDispatch; ASSERT_EQ(add.emoteSetID, EMOTE_SET_A); @@ -73,7 +75,7 @@ TEST(SeventvEventAPI, AllEvents) removeDispatch = std::nullopt; eventAPI.subscribeUser(TARGET_USER_ID, ""); - std::this_thread::sleep_for(50ms); + QTest::qWait(50); ASSERT_EQ(addDispatch.has_value(), false); ASSERT_EQ(updateDispatch.has_value(), false); @@ -87,23 +89,26 @@ TEST(SeventvEventAPI, AllEvents) ASSERT_EQ(user.connectionIndex, 0); eventAPI.stop(); - ASSERT_EQ(eventAPI.diag.connectionsOpened, 1); - ASSERT_EQ(eventAPI.diag.connectionsClosed, 1); - ASSERT_EQ(eventAPI.diag.connectionsFailed, 0); + // after exactly one event loop iteration, we should see updated counters + QCoreApplication::processEvents(QEventLoop::AllEvents); + QCoreApplication::sendPostedEvents(nullptr, QEvent::DeferredDelete); + + ASSERT_EQ(eventAPI.diag().connectionsOpened, 1); + ASSERT_EQ(eventAPI.diag().connectionsClosed, 1); + ASSERT_EQ(eventAPI.diag().connectionsFailed, 0); } TEST(SeventvEventAPI, NoHeartbeat) { + mock::BaseApplication app; const QString host("wss://127.0.0.1:9050/liveupdates/seventv/no-heartbeat"); SeventvEventAPI eventApi(host, std::chrono::milliseconds(1000)); - eventApi.start(); - std::this_thread::sleep_for(50ms); eventApi.subscribeUser("", EMOTE_SET_A); - std::this_thread::sleep_for(1250ms); - ASSERT_EQ(eventApi.diag.connectionsOpened, 2); - ASSERT_EQ(eventApi.diag.connectionsClosed, 1); - ASSERT_EQ(eventApi.diag.connectionsFailed, 0); + QTest::qWait(1250); + ASSERT_EQ(eventApi.diag().connectionsOpened, 2); + ASSERT_EQ(eventApi.diag().connectionsClosed, 1); + ASSERT_EQ(eventApi.diag().connectionsFailed, 0); eventApi.stop(); }