diff --git a/CHANGELOG.md b/CHANGELOG.md index 495b4734..f861f18a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,7 +65,7 @@ - Dev: Added Qt keyword and warning flags project wide. (#6520) - Dev: Added REUSE license checking. (#6659) - Def: Fixed compilation error in tests with Clang 21. (#6519) -- Dev: The 7TV and BTTV liveupdates now use Boost.Beast's WebSockets. (#6308) +- Dev: The 7TV and BTTV liveupdates now use Boost.Beast's WebSockets. (#6308, #6763) - 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/providers/liveupdates/BasicPubSubManager.hpp b/src/providers/liveupdates/BasicPubSubManager.hpp index bf0ed683..bdf1e5f4 100644 --- a/src/providers/liveupdates/BasicPubSubManager.hpp +++ b/src/providers/liveupdates/BasicPubSubManager.hpp @@ -107,6 +107,8 @@ public: protected: void unsubscribe(const Subscription &subscription) { + assertInGuiThread(); + for (auto &client : this->clients_) { if (client.second->unsubscribe(subscription)) @@ -118,6 +120,8 @@ protected: void subscribe(const Subscription &subscription) { + assertInGuiThread(); + if (this->trySubscribe(subscription)) { return; @@ -141,6 +145,8 @@ private: void onConnectionOpen(size_t id) { + assertInGuiThread(); + DebugCount::increase("LiveUpdates connections"); this->addingClient_ = false; this->diag.connectionsOpened.fetch_add(1, std::memory_order_acq_rel); @@ -164,8 +170,8 @@ private: { qCDebug(chatterinoLiveupdates) << "Failed to subscribe to" << last << "on new client."; - // TODO: should we try to add a new client here? - return; + this->pendingSubscriptions_.emplace_back(std::move(last)); + break; } DebugCount::decrease("LiveUpdates subscription backlog"); pendingSubsToTake--; @@ -173,12 +179,17 @@ private: if (!this->pendingSubscriptions_.empty()) { + qCDebug(chatterinoLiveupdates) + << "Adding another client for " + << this->pendingSubscriptions_.size() << "subs"; this->addClient(); } } void onConnectionClose(size_t id) { + assertInGuiThread(); + this->addingClient_ = false; auto it = this->clients_.find(id); @@ -216,10 +227,14 @@ private: { qCWarning(chatterinoLiveupdates) << "Retrying after" << id << "failed"; + auto nSubs = subs.size(); + DebugCount::increase("LiveUpdates subscription backlog", + static_cast(nSubs)); 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(); }); @@ -234,6 +249,8 @@ private: void addClient() { + assertInGuiThread(); + if (this->addingClient_ || !this->pool_) { return; diff --git a/tests/src/BasicPubSub.cpp b/tests/src/BasicPubSub.cpp index caae11df..e9b66f46 100644 --- a/tests/src/BasicPubSub.cpp +++ b/tests/src/BasicPubSub.cpp @@ -80,8 +80,9 @@ class MyManager; class MyClient : public BasicPubSubClient { public: - MyClient(MyManager &manager) - : manager(manager) + MyClient(MyManager &manager, size_t limit) + : BasicPubSubClient(limit) + , manager(manager) { } @@ -94,8 +95,9 @@ private: class MyManager : public BasicPubSubManager { public: - MyManager(QString host) + MyManager(QString host, size_t limit = 100) : BasicPubSubManager(std::move(host), "Test") + , limit_(limit) { } @@ -126,12 +128,13 @@ public: std::shared_ptr makeClient() { - return std::make_shared(*this); + return std::make_shared(*this, this->limit_); } private: std::mutex messageMtx_; std::deque messageQueue_; + size_t limit_; friend MyClient; }; @@ -177,3 +180,46 @@ TEST(BasicPubSub, SubscriptionCycle) ASSERT_EQ(manager.diag.connectionsFailed, 0); ASSERT_EQ(manager.messagesReceived, 2); } + +TEST(BasicPubSub, SubLimits) +{ + mock::BaseApplication app; + const QString host("wss://127.0.0.1:9050/liveupdates/sub-unsub"); + MyManager manager(host, 1); + manager.sub({.type = 1, .condition = "foo"}); + manager.sub({.type = 2, .condition = "foo"}); + manager.sub({.type = 3, .condition = "foo"}); + manager.sub({.type = 4, .condition = "foo"}); + manager.sub({.type = 5, .condition = "foo"}); + QTest::qWait(500); + ASSERT_EQ(manager.diag.connectionsOpened, 5); + ASSERT_EQ(manager.diag.connectionsClosed, 0); + ASSERT_EQ(manager.diag.connectionsFailed, 0); + ASSERT_EQ(manager.messagesReceived, 5); + + // The messages come from multiple connections, so they're not necessarily + // in order. + std::vector messages; + for (size_t i = 0; i < 5; i++) + { + auto msg = manager.popMessage(); + ASSERT_TRUE(msg.has_value()); + messages.push_back(*std::move(msg)); + } + std::ranges::sort(messages); + ASSERT_EQ(messages[0], QString("ack-sub-1-foo")); + ASSERT_EQ(messages[1], QString("ack-sub-2-foo")); + ASSERT_EQ(messages[2], QString("ack-sub-3-foo")); + ASSERT_EQ(messages[3], QString("ack-sub-4-foo")); + ASSERT_EQ(messages[4], QString("ack-sub-5-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, 5); + ASSERT_EQ(manager.diag.connectionsClosed, 5); + ASSERT_EQ(manager.diag.connectionsFailed, 0); + ASSERT_EQ(manager.messagesReceived, 5); +}