Close and flush the previous player on reconnect in egress

The listener factory overwrote impl->player without closing the old
session, so a stalled player replaced by a reconnect left its fd open
forever: the writer thread stayed blocked in the old player's send(),
writer.join() in stop() never returned, and the fd that would unblock
it was only closed after that join. Shut down the outgoing session's
fd under player_mutex before installing the new one so a blocked
writer unblocks.

Also add PlayerQueue::clear() and call it in the same critical
section, after closing the old player and before installing the new
one, so coded frames queued during the previous player's tenure are
dropped instead of sitting ahead of the new player's init frames.
clear() does not count toward dropped() -- it is a deliberate flush,
not the loss dropped() tracks. Add a unit test covering that.

Author: mondain

gateway/include/roqr/gateway/player_queue.hpp

@@ -36,6 +36,12 @@ class PlayerQueue {
     size_t size() const;
     uint64_t dropped() const;
 
+    // Drop all queued entries without counting them toward dropped() — used
+    // when installing a new player so stale prior-era frames don't linger
+    // ahead of the new player's init frames; this is a deliberate flush, not
+    // the loss dropped() tracks.
+    void clear();
+
 private:
     struct Entry {
         roqr::rtmp::RtmpMessage msg;

gateway/src/egress.cpp

@@ -158,6 +158,14 @@ bool EgressGateway::start(const EgressOptions& options) {
         [impl](roqr::rtmp::ServerSession& s) {
             {
                 std::lock_guard lock(impl->player_mutex);
+                // A previous player may still be alive (e.g. a stalled one
+                // that never disconnected): shut its fd down so any writer
+                // blocked in its send() unblocks, then drop its queued
+                // frames so the new player isn't fed stale, prior-era media
+                // ahead of its own init frames.
+                if (impl->player != nullptr && impl->player != &s)
+                    impl->player->close();
+                impl->queue.clear();
                 impl->player = &s;
                 impl->player_ready = false;
             }

gateway/src/player_queue.cpp

@@ -56,4 +56,9 @@ uint64_t PlayerQueue::dropped() const {
     return dropped_;
 }
 
+void PlayerQueue::clear() {
+    std::lock_guard lock(mutex_);
+    queue_.clear();
+}
+
 }  // namespace roqr::gateway

tests/gateway/player_queue_test.cpp

@@ -76,3 +76,21 @@ TEST_CASE("close unblocks a waiting consumer and drains remaining") {
     q2.close();
     consumer.join();
 }
+
+TEST_CASE("clear drops all entries without counting them as dropped") {
+    PlayerQueue q(8);
+    REQUIRE(q.push(seq_header(), PlayerQueue::Kind::Init));
+    REQUIRE(q.push(vid(1), PlayerQueue::Kind::Coded));
+    REQUIRE(q.push(vid(2), PlayerQueue::Kind::Coded));
+    CHECK(q.size() == 3);
+    CHECK(q.dropped() == 0);
+
+    q.clear();
+
+    CHECK(q.size() == 0);
+    CHECK(q.dropped() == 0);  // clear() is not a drop-tracked eviction
+
+    // Queue is still usable after clear().
+    REQUIRE(q.push(vid(3), PlayerQueue::Kind::Coded));
+    CHECK(q.pop()->timestamp == 3);
+}