Process incoming Unsubscribes.
Not in production.
PiperOrigin-RevId: 606667410
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index 0c425a1..11feb9e 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -643,6 +643,13 @@
session_->active_subscribes_.erase(it);
}
+void MoqtSession::Stream::OnUnsubscribeMessage(const MoqtUnsubscribe& message) {
+ // Search all the tracks to find the subscribe ID.
+ for (auto& [name, track] : session_->local_tracks_) {
+ track.DeleteWindow(message.subscribe_id);
+ }
+}
+
void MoqtSession::Stream::OnAnnounceMessage(const MoqtAnnounce& message) {
if (!CheckIfIsControlStream()) {
return;
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index 9b77a11..911849a 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h
@@ -157,7 +157,7 @@
void OnSubscribeMessage(const MoqtSubscribe& message) override;
void OnSubscribeOkMessage(const MoqtSubscribeOk& message) override;
void OnSubscribeErrorMessage(const MoqtSubscribeError& message) override;
- void OnUnsubscribeMessage(const MoqtUnsubscribe& /*message*/) override {}
+ void OnUnsubscribeMessage(const MoqtUnsubscribe& message) override;
void OnSubscribeFinMessage(const MoqtSubscribeFin& /*message*/) override {}
void OnSubscribeRstMessage(const MoqtSubscribeRst& /*message*/) override {}
void OnAnnounceMessage(const MoqtAnnounce& message) override;
@@ -231,6 +231,7 @@
MoqtSubscribe message;
RemoteTrack::Visitor* visitor;
};
+ // Outgoing SUBSCRIBEs that have not received SUBSCRIBE_OK or SUBSCRIBE_ERROR.
absl::flat_hash_map<uint64_t, ActiveSubscribe> active_subscribes_;
uint64_t next_subscribe_id_ = 0;
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc
index 1f4f9c0..a865118 100644
--- a/quiche/quic/moqt/moqt_session_test.cc
+++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -871,6 +871,22 @@
EXPECT_TRUE(reported_error);
}
+TEST_F(MoqtSessionTest, ReceiveUnsubscribe) {
+ FullTrackName ftn("foo", "bar");
+ MockLocalTrackVisitor visitor;
+ session_.AddLocalTrack(ftn, &visitor);
+ MoqtSessionPeer::AddSubscription(&session_, ftn, 0, 1, 3, 4);
+ EXPECT_TRUE(session_.HasSubscribers(ftn));
+ StrictMock<webtransport::test::MockStream> mock_stream;
+ std::unique_ptr<MoqtParserVisitor> stream_input =
+ MoqtSessionPeer::CreateControlStream(&session_, &mock_stream);
+ MoqtUnsubscribe unsubscribe = {
+ /*subscribe_id=*/0,
+ };
+ stream_input->OnUnsubscribeMessage(unsubscribe);
+ EXPECT_FALSE(session_.HasSubscribers(ftn));
+}
+
// TODO: Cover more error cases in the above
} // namespace test
diff --git a/quiche/quic/moqt/moqt_subscribe_windows.cc b/quiche/quic/moqt/moqt_subscribe_windows.cc
index 335d2d8..9bea570 100644
--- a/quiche/quic/moqt/moqt_subscribe_windows.cc
+++ b/quiche/quic/moqt/moqt_subscribe_windows.cc
@@ -96,9 +96,9 @@
std::vector<SubscribeWindow*> MoqtSubscribeWindows::SequenceIsSubscribed(
FullSequence sequence) {
std::vector<SubscribeWindow*> retval;
- for (auto it = windows.begin(); it != windows.end(); it++) {
- if (it->InWindow(sequence)) {
- retval.push_back(&(*it));
+ for (auto& [subscribe_id, window] : windows_) {
+ if (window.InWindow(sequence)) {
+ retval.push_back(&(window));
}
}
return retval;
diff --git a/quiche/quic/moqt/moqt_subscribe_windows.h b/quiche/quic/moqt/moqt_subscribe_windows.h
index f8def02..e6ec05d 100644
--- a/quiche/quic/moqt/moqt_subscribe_windows.h
+++ b/quiche/quic/moqt/moqt_subscribe_windows.h
@@ -6,11 +6,10 @@
#define QUICHE_QUIC_MOQT_SUBSCRIBE_WINDOWS_H
#include <cstdint>
-#include <list>
#include <optional>
#include <vector>
-#include "absl/container/flat_hash_map.h"
+#include "absl/container/node_hash_map.h"
#include "quiche/quic/moqt/moqt_messages.h"
#include "quiche/common/platform/api/quiche_export.h"
#include "quiche/web_transport/web_transport.h"
@@ -77,12 +76,24 @@
// |window| has already been converted into absolute sequence numbers. An
// optimization could consolidate overlapping subscribe windows.
- void AddWindow(SubscribeWindow window) { windows.push_front(window); }
+ void AddWindow(SubscribeWindow window) {
+ windows_.emplace(window.subscribe_id(), window);
+ }
+ void RemoveWindow(uint64_t subscribe_id) { windows_.erase(subscribe_id); }
- bool IsEmpty() const { return windows.empty(); }
+ bool IsEmpty() const { return windows_.empty(); }
+
+ SubscribeWindow* GetWindow(uint64_t subscribe_id) {
+ auto it = windows_.find(subscribe_id);
+ if (it == windows_.end()) {
+ return nullptr;
+ }
+ return &it->second;
+ }
private:
- std::list<SubscribeWindow> windows;
+ // Indexed by Subscribe ID.
+ absl::node_hash_map<uint64_t, SubscribeWindow> windows_;
};
} // namespace moqt
diff --git a/quiche/quic/moqt/moqt_subscribe_windows_test.cc b/quiche/quic/moqt/moqt_subscribe_windows_test.cc
index 02e7007..8387a7e 100644
--- a/quiche/quic/moqt/moqt_subscribe_windows_test.cc
+++ b/quiche/quic/moqt/moqt_subscribe_windows_test.cc
@@ -166,6 +166,15 @@
EXPECT_EQ(hits[1]->subscribe_id(), 0);
}
+TEST_F(MoqtSubscribeWindowsTest, AddGetRemoveWindow) {
+ windows_.AddWindow(SubscribeWindow(0, 1, 0, 3, 9));
+ SubscribeWindow* window = windows_.GetWindow(0);
+ EXPECT_EQ(window->subscribe_id(), 0);
+ EXPECT_EQ(windows_.GetWindow(1), nullptr);
+ windows_.RemoveWindow(0);
+ EXPECT_EQ(windows_.GetWindow(0), nullptr);
+}
+
} // namespace test
} // namespace moqt
diff --git a/quiche/quic/moqt/moqt_track.h b/quiche/quic/moqt/moqt_track.h
index 4d04afc..e52b084 100644
--- a/quiche/quic/moqt/moqt_track.h
+++ b/quiche/quic/moqt/moqt_track.h
@@ -56,6 +56,9 @@
}
void AddWindow(SubscribeWindow window) { windows_.AddWindow(window); }
+ void DeleteWindow(uint64_t subscribe_id) {
+ windows_.RemoveWindow(subscribe_id);
+ }
// Returns the largest observed sequence, but increments the object sequence
// by one.
@@ -70,6 +73,10 @@
bool HasSubscriber() const { return !windows_.IsEmpty(); }
+ SubscribeWindow* GetWindow(uint64_t subscribe_id) {
+ return windows_.GetWindow(subscribe_id);
+ }
+
private:
// This only needs to track subscriptions to current and future objects;
// requests for objects in the past are forwarded to the application.
diff --git a/quiche/quic/moqt/moqt_track_test.cc b/quiche/quic/moqt/moqt_track_test.cc
index 4d85b6b..46fd3c5 100644
--- a/quiche/quic/moqt/moqt_track_test.cc
+++ b/quiche/quic/moqt/moqt_track_test.cc
@@ -42,6 +42,14 @@
EXPECT_EQ(track_.track_alias(), 6);
}
+TEST_F(LocalTrackTest, AddGetDeleteWindow) {
+ track_.AddWindow(SubscribeWindow(0, 4, 1));
+ EXPECT_EQ(track_.GetWindow(0)->subscribe_id(), 0);
+ EXPECT_EQ(track_.GetWindow(1), nullptr);
+ track_.DeleteWindow(0);
+ EXPECT_EQ(track_.GetWindow(0), nullptr);
+}
+
TEST_F(LocalTrackTest, ShouldSend) {
track_.AddWindow(SubscribeWindow(0, 4, 1));
EXPECT_TRUE(track_.HasSubscriber());
