Update MoQ chat to draft-00.
Use of a CATALOG was replaced with SUBSCRIBE_ANNOUNCES and ANNOUNCE.
Minor changes to MoQT session based on revealed suboptimal code.
Successful interop with Meta.
PiperOrigin-RevId: 712526076
diff --git a/build/source_list.bzl b/build/source_list.bzl
index a288134..32e4def 100644
--- a/build/source_list.bzl
+++ b/build/source_list.bzl
@@ -1574,6 +1574,7 @@
"quic/moqt/tools/chat_client_bin.cc",
"quic/moqt/tools/chat_server.cc",
"quic/moqt/tools/chat_server_bin.cc",
+ "quic/moqt/tools/moq_chat.cc",
"quic/moqt/tools/moq_chat_end_to_end_test.cc",
"quic/moqt/tools/moq_chat_test.cc",
"quic/moqt/tools/moqt_client.cc",
diff --git a/build/source_list.gni b/build/source_list.gni
index 52927da..3b75a6b 100644
--- a/build/source_list.gni
+++ b/build/source_list.gni
@@ -1578,6 +1578,7 @@
"src/quiche/quic/moqt/tools/chat_client_bin.cc",
"src/quiche/quic/moqt/tools/chat_server.cc",
"src/quiche/quic/moqt/tools/chat_server_bin.cc",
+ "src/quiche/quic/moqt/tools/moq_chat.cc",
"src/quiche/quic/moqt/tools/moq_chat_end_to_end_test.cc",
"src/quiche/quic/moqt/tools/moq_chat_test.cc",
"src/quiche/quic/moqt/tools/moqt_client.cc",
diff --git a/build/source_list.json b/build/source_list.json
index d1b698f..5eb9364 100644
--- a/build/source_list.json
+++ b/build/source_list.json
@@ -1577,6 +1577,7 @@
"quiche/quic/moqt/tools/chat_client_bin.cc",
"quiche/quic/moqt/tools/chat_server.cc",
"quiche/quic/moqt/tools/chat_server_bin.cc",
+ "quiche/quic/moqt/tools/moq_chat.cc",
"quiche/quic/moqt/tools/moq_chat_end_to_end_test.cc",
"quiche/quic/moqt/tools/moq_chat_test.cc",
"quiche/quic/moqt/tools/moqt_client.cc",
diff --git a/quiche/quic/moqt/moqt_integration_test.cc b/quiche/quic/moqt/moqt_integration_test.cc
index a268578..cb8a44e 100644
--- a/quiche/quic/moqt/moqt_integration_test.cc
+++ b/quiche/quic/moqt/moqt_integration_test.cc
@@ -2,7 +2,6 @@
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
-#include <cstdint>
#include <memory>
#include <optional>
#include <string>
@@ -22,7 +21,6 @@
#include "quiche/quic/moqt/test_tools/moqt_simulator_harness.h"
#include "quiche/quic/moqt/tools/moqt_mock_visitor.h"
#include "quiche/quic/test_tools/quic_test_utils.h"
-#include "quiche/quic/test_tools/simulator/simulator.h"
#include "quiche/quic/test_tools/simulator/test_harness.h"
#include "quiche/common/platform/api/quiche_test.h"
diff --git a/quiche/quic/moqt/moqt_messages.h b/quiche/quic/moqt/moqt_messages.h
index 508d103..f898d01 100644
--- a/quiche/quic/moqt/moqt_messages.h
+++ b/quiche/quic/moqt/moqt_messages.h
@@ -159,6 +159,7 @@
enum class MoqtAnnounceErrorCode : uint64_t {
kInternalError = 0,
kAnnounceNotSupported = 1,
+ kNotASubscribedNamespace = 2,
};
enum class QUICHE_EXPORT SubscribeErrorCode : uint64_t {
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index ddc9ae9..f46e655 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -422,6 +422,7 @@
if (track == nullptr) {
return;
}
+ QUIC_DLOG(INFO) << ENDPOINT << "Sent UNSUBSCRIBE message for " << name;
MoqtUnsubscribe message;
message.subscribe_id = track->subscribe_id();
SendControlMessage(framer_.SerializeUnsubscribe(message));
@@ -525,7 +526,7 @@
subscribe_done.final_id = subscription.largest_sent();
SendControlMessage(framer_.SerializeSubscribeDone(subscribe_done));
QUIC_DLOG(INFO) << ENDPOINT << "Sent SUBSCRIBE_DONE message for "
- << subscribe_id;
+ << subscription.publisher().GetTrackName();
// Clean up the subscription
published_subscriptions_.erase(it);
for (webtransport::StreamId stream_id : streams_to_reset) {
@@ -1007,6 +1008,12 @@
void MoqtSession::ControlStream::OnUnsubscribeMessage(
const MoqtUnsubscribe& message) {
+ auto it = session_->published_subscriptions_.find(message.subscribe_id);
+ if (it == session_->published_subscriptions_.end()) {
+ return;
+ }
+ QUIC_DLOG(INFO) << ENDPOINT << "Received an UNSUBSCRIBE for "
+ << it->second->publisher().GetTrackName();
session_->SubscribeIsDone(message.subscribe_id,
SubscribeDoneCode::kUnsubscribed, "");
}
diff --git a/quiche/quic/moqt/tools/chat_client.cc b/quiche/quic/moqt/tools/chat_client.cc
index e98cf75..fa0c7db 100644
--- a/quiche/quic/moqt/tools/chat_client.cc
+++ b/quiche/quic/moqt/tools/chat_client.cc
@@ -7,17 +7,14 @@
#include <poll.h>
#include <unistd.h>
-#include <cstdint>
#include <iostream>
#include <memory>
#include <optional>
-#include <sstream>
#include <string>
#include <utility>
-#include <vector>
#include "absl/container/flat_hash_map.h"
-#include "absl/strings/str_split.h"
+#include "absl/functional/bind_front.h"
#include "absl/strings/string_view.h"
#include "quiche/quic/core/crypto/proof_verifier.h"
#include "quiche/quic/core/io/quic_default_event_loop.h"
@@ -29,7 +26,7 @@
#include "quiche/quic/moqt/moqt_outgoing_queue.h"
#include "quiche/quic/moqt/moqt_priority.h"
#include "quiche/quic/moqt/moqt_session.h"
-#include "quiche/quic/moqt/moqt_track.h"
+#include "quiche/quic/moqt/tools/moq_chat.h"
#include "quiche/quic/moqt/tools/moqt_client.h"
#include "quiche/quic/platform/api/quic_default_proof_providers.h"
#include "quiche/quic/platform/api/quic_socket_address.h"
@@ -39,13 +36,60 @@
#include "quiche/common/quiche_buffer_allocator.h"
#include "quiche/common/simple_buffer_allocator.h"
-namespace moqt {
+namespace moqt::moq_chat {
+
+std::optional<MoqtAnnounceErrorReason> ChatClient::OnIncomingAnnounce(
+ const moqt::FullTrackName& track_namespace, AnnounceEvent announce_type) {
+ if (track_namespace == GetUserNamespace(my_track_name_)) {
+ // Ignore ANNOUNCE for my own track.
+ return std::optional<MoqtAnnounceErrorReason>();
+ }
+ std::optional<FullTrackName> track_name = ConstructTrackNameFromNamespace(
+ track_namespace, GetChatId(my_track_name_));
+ if (announce_type == AnnounceEvent::kUnannounce) {
+ std::cout << "UNANNOUNCE for " << track_namespace.ToString() << "\n";
+ if (track_name.has_value() && other_users_.contains(*track_name)) {
+ session_->Unsubscribe(*track_name);
+ other_users_.erase(*track_name);
+ }
+ return std::nullopt;
+ }
+ std::cout << "ANNOUNCE for " << track_namespace.ToString() << "\n";
+ if (!track_name.has_value()) {
+ std::cout << "ANNOUNCE rejected, invalid namespace\n";
+ return std::make_optional<MoqtAnnounceErrorReason>(
+ MoqtAnnounceErrorCode::kNotASubscribedNamespace,
+ "Not a subscribed namespace");
+ }
+ if (other_users_.contains(*track_name)) {
+ std::cout << "Duplicate ANNOUNCE, send OK and ignore\n";
+ return std::nullopt;
+ }
+ if (GetUsername(my_track_name_) == GetUsername(*track_name)) {
+ std::cout << "ANNOUNCE for a previous instance of my track, "
+ "do not subscribe\n";
+ return std::nullopt;
+ }
+ if (session_->SubscribeCurrentGroup(
+ *track_name, &remote_track_visitor_,
+ MoqtSubscribeParameters{std::string(GetUsername(my_track_name_)),
+ std::nullopt, std::nullopt, std::nullopt})) {
+ ++subscribes_to_make_;
+ other_users_.emplace(*track_name);
+ }
+ return std::nullopt; // Send ANNOUNCE_OK.
+}
ChatClient::ChatClient(const quic::QuicServerId& server_id,
bool ignore_certificate,
std::unique_ptr<ChatUserInterface> interface,
+ absl::string_view chat_id, absl::string_view username,
+ absl::string_view localhost,
quic::QuicEventLoop* event_loop)
- : event_loop_(event_loop), interface_(std::move(interface)) {
+ : my_track_name_(ConstructTrackName(chat_id, username, localhost)),
+ event_loop_(event_loop),
+ remote_track_visitor_(this),
+ interface_(std::move(interface)) {
if (event_loop_ == nullptr) {
quic::QuicDefaultClock* clock = quic::QuicDefaultClock::Get();
local_event_loop_ = quic::GetDefaultEventLoop()->Create(clock);
@@ -72,10 +116,13 @@
std::cerr << "Closed session, reason = " << error_message << "\n";
session_is_open_ = false;
connect_failed_ = true;
+ session_ = nullptr;
};
session_callbacks_.session_deleted_callback = [this]() {
session_ = nullptr;
};
+ session_callbacks_.incoming_announce_callback =
+ absl::bind_front(&ChatClient::OnIncomingAnnounce, this);
interface_->Initialize(
[this](absl::string_view input_message) {
OnTerminalLineInput(input_message);
@@ -83,10 +130,7 @@
event_loop_);
}
-bool ChatClient::Connect(absl::string_view path, absl::string_view username,
- absl::string_view chat_id) {
- username_ = username;
- chat_strings_.emplace(chat_id);
+bool ChatClient::Connect(absl::string_view path) {
client_->Connect(std::string(path), std::move(session_callbacks_));
while (!session_is_open_ && !connect_failed_) {
RunEventLoop();
@@ -99,6 +143,13 @@
return;
}
if (input_message == "/exit") {
+ // Clean teardown of SUBSCRIBE_ANNOUNCES, ANNOUNCE, SUBSCRIBE.
+ session_->UnsubscribeAnnounces(GetChatNamespace(my_track_name_));
+ session_->Unannounce(GetUserNamespace(my_track_name_));
+ for (const auto& track_name : other_users_) {
+ session_->Unsubscribe(track_name);
+ }
+ other_users_.clear();
session_is_open_ = false;
return;
}
@@ -111,175 +162,96 @@
const FullTrackName& full_track_name,
std::optional<FullSequence> /*largest_id*/,
std::optional<absl::string_view> reason_phrase) {
- client_->subscribes_to_make_--;
- if (full_track_name == client_->chat_strings_->GetCatalogName()) {
- std::cout << "Subscription to catalog ";
- } else {
- std::cout << "Subscription to user " << full_track_name.ToString() << " ";
+ auto it = client_->other_users_.find(full_track_name);
+ if (it == client_->other_users_.end()) {
+ std::cout << "Error: received reply for unknown user "
+ << full_track_name.ToString() << "\n";
+ return;
}
+ --client_->subscribes_to_make_;
+ std::cout << "Subscription to user " << GetUsername(*it) << " ";
if (reason_phrase.has_value()) {
std::cout << "REJECTED, reason = " << *reason_phrase << "\n";
+ client_->other_users_.erase(it);
} else {
std::cout << "ACCEPTED\n";
}
}
void ChatClient::RemoteTrackVisitor::OnObjectFragment(
- const FullTrackName& full_track_name, FullSequence sequence,
+ const FullTrackName& full_track_name, FullSequence /*sequence*/,
MoqtPriority /*publisher_priority*/, MoqtObjectStatus /*status*/,
absl::string_view object, bool end_of_message) {
if (!end_of_message) {
std::cerr << "Error: received partial message despite requesting "
"buffering\n";
}
- if (full_track_name == client_->chat_strings_->GetCatalogName()) {
- if (sequence.group < client_->catalog_group_) {
- std::cout << "Ignoring old catalog";
- return;
- }
- client_->ProcessCatalog(object, this, sequence.group, sequence.object);
- return;
- }
- std::string username(
- client_->chat_strings_->GetUsernameFromFullTrackName(full_track_name));
- if (!client_->other_users_.contains(username)) {
- std::cout << "Username " << username << "doesn't exist\n";
+ auto it = client_->other_users_.find(full_track_name);
+ if (it == client_->other_users_.end()) {
+ std::cout << "Error: received message for unknown user "
+ << full_track_name.ToString() << "\n";
return;
}
if (object.empty()) {
return;
}
- client_->WriteToOutput(username, object);
+ client_->WriteToOutput(GetUsername(*it), object);
}
-bool ChatClient::AnnounceAndSubscribe() {
+bool ChatClient::AnnounceAndSubscribeAnnounces() {
session_ = client_->session();
if (session_ == nullptr) {
std::cout << "Failed to connect.\n";
return false;
}
- if (!username_.empty()) {
- // A server log might choose to not provide a username, thus getting all
- // the messages without adding itself to the catalog.
- FullTrackName my_track_name =
- chat_strings_->GetFullTrackNameFromUsername(username_);
- queue_ = std::make_shared<MoqtOutgoingQueue>(
- my_track_name, MoqtForwardingPreference::kSubgroup);
- publisher_.Add(queue_);
- session_->set_publisher(&publisher_);
- MoqtOutgoingAnnounceCallback announce_callback =
- [this](FullTrackName track_namespace,
- std::optional<MoqtAnnounceErrorReason> reason) {
- if (reason.has_value()) {
- std::cout << "ANNOUNCE rejected, " << reason->reason_phrase << "\n";
- session_->Error(MoqtError::kInternalError,
- "Local ANNOUNCE rejected");
- return;
- }
- std::cout << "ANNOUNCE for " << track_namespace.ToString()
- << " accepted\n";
+ // TODO: A server log might choose to not provide a username, thus getting all
+ // the messages without adding itself to the catalog.
+ queue_ = std::make_shared<MoqtOutgoingQueue>(
+ my_track_name_, MoqtForwardingPreference::kSubgroup);
+ publisher_.Add(queue_);
+ session_->set_publisher(&publisher_);
+ MoqtOutgoingAnnounceCallback announce_callback =
+ [this](FullTrackName track_namespace,
+ std::optional<MoqtAnnounceErrorReason> reason) {
+ if (reason.has_value()) {
+ std::cout << "ANNOUNCE rejected, " << reason->reason_phrase << "\n";
+ session_->Error(MoqtError::kInternalError, "Local ANNOUNCE rejected");
return;
- };
- FullTrackName my_track_namespace = my_track_name;
- my_track_namespace.NameToNamespace();
- std::cout << "Announcing " << my_track_namespace.ToString() << "\n";
- session_->Announce(my_track_namespace, std::move(announce_callback));
- }
- remote_track_visitor_ = std::make_unique<RemoteTrackVisitor>(this);
- FullTrackName catalog_name = chat_strings_->GetCatalogName();
- if (!session_->SubscribeCurrentGroup(
- catalog_name, remote_track_visitor_.get(),
- MoqtSubscribeParameters{username_, std::nullopt, std::nullopt,
- std::nullopt})) {
- std::cout << "Failed to get catalog\n";
- return false;
- }
+ }
+ std::cout << "ANNOUNCE for " << track_namespace.ToString()
+ << " accepted\n";
+ return;
+ };
+ std::cout << "Announcing " << GetUserNamespace(my_track_name_).ToString()
+ << "\n";
+ session_->Announce(GetUserNamespace(my_track_name_),
+ std::move(announce_callback));
+
+ // Send SUBSCRIBE_ANNOUNCE. Pop 3 levels of namespace to get to {moq-chat,
+ // chat-id}
+ MoqtOutgoingSubscribeAnnouncesCallback subscribe_announces_callback =
+ [this](FullTrackName track_namespace,
+ std::optional<SubscribeErrorCode> error,
+ absl::string_view reason) {
+ if (error.has_value()) {
+ std::cout << "SUBSCRIBE_ANNOUNCES rejected, " << reason << "\n";
+ session_->Error(MoqtError::kInternalError,
+ "Local SUBSCRIBE_ANNOUNCES rejected");
+ return;
+ }
+ std::cout << "SUBSCRIBE_ANNOUNCES for " << track_namespace.ToString()
+ << " accepted\n";
+ return;
+ };
+ session_->SubscribeAnnounces(
+ GetChatNamespace(my_track_name_), std::move(subscribe_announces_callback),
+ MoqtSubscribeParameters{std::string(GetUsername(my_track_name_)),
+ std::nullopt, std::nullopt, std::nullopt});
+
while (session_is_open_ && is_syncing()) {
RunEventLoop();
}
return session_is_open_;
}
-void ChatClient::ProcessCatalog(absl::string_view object,
- SubscribeRemoteTrack::Visitor* visitor,
- uint64_t group_sequence,
- uint64_t object_sequence) {
- std::string message(object);
- std::istringstream f(message);
- // std::string line;
- bool got_version = true;
- if (object_sequence == 0) {
- std::cout << "Received new Catalog. Users:\n";
- got_version = false;
- }
- std::vector<absl::string_view> lines =
- absl::StrSplit(object, '\n', absl::SkipEmpty());
- for (absl::string_view line : lines) {
- if (!got_version) {
- if (line != "version=1") {
- session_->Error(MoqtError::kProtocolViolation,
- "Catalog does not begin with version");
- return;
- }
- got_version = true;
- continue;
- }
- std::string user;
- bool add = true;
- if (object_sequence > 0) {
- switch (line[0]) {
- case '-':
- add = false;
- break;
- case '+':
- break;
- default:
- std::cerr << "Catalog update with neither + nor -\n";
- return;
- }
- user = line.substr(1, line.size() - 1);
- } else {
- user = line;
- }
- if (username_ == user) {
- std::cout << user << "\n";
- continue;
- }
- if (!add) {
- session_->Unsubscribe(chat_strings_->GetFullTrackNameFromUsername(user));
- std::cout << user << "left the chat\n";
- other_users_.erase(user);
- continue;
- }
- if (object_sequence == 0) {
- std::cout << user << "\n";
- } else {
- std::cout << user << " joined the chat\n";
- }
- auto it = other_users_.find(user);
- if (it == other_users_.end()) {
- FullTrackName to_subscribe =
- chat_strings_->GetFullTrackNameFromUsername(user);
- auto new_user = other_users_.emplace(
- std::make_pair(user, ChatUser(to_subscribe, group_sequence)));
- ChatUser& user_record = new_user.first->second;
- session_->SubscribeCurrentGroup(user_record.full_track_name, visitor);
- subscribes_to_make_++;
- } else {
- if (it->second.from_group == group_sequence) {
- session_->Error(MoqtError::kProtocolViolation,
- "User listed twice in Catalog");
- return;
- }
- it->second.from_group = group_sequence;
- }
- }
- if (object_sequence == 0) { // Eliminate users that are no longer present
- absl::erase_if(other_users_, [&](const auto& kv) {
- return kv.second.from_group != group_sequence;
- });
- }
- catalog_group_ = group_sequence;
-}
-
-} // namespace moqt
+} // namespace moqt::moq_chat
diff --git a/quiche/quic/moqt/tools/chat_client.h b/quiche/quic/moqt/tools/chat_client.h
index e2c341f..b56142c 100644
--- a/quiche/quic/moqt/tools/chat_client.h
+++ b/quiche/quic/moqt/tools/chat_client.h
@@ -5,12 +5,10 @@
#ifndef QUICHE_QUIC_MOQT_TOOLS_CHAT_CLIENT_H
#define QUICHE_QUIC_MOQT_TOOLS_CHAT_CLIENT_H
-#include <cstdint>
#include <memory>
#include <optional>
-#include <string>
-#include "absl/container/flat_hash_map.h"
+#include "absl/container/flat_hash_set.h"
#include "absl/strings/string_view.h"
#include "quiche/quic/core/io/quic_event_loop.h"
#include "quiche/quic/core/quic_server_id.h"
@@ -21,12 +19,11 @@
#include "quiche/quic/moqt/moqt_priority.h"
#include "quiche/quic/moqt/moqt_session.h"
#include "quiche/quic/moqt/moqt_track.h"
-#include "quiche/quic/moqt/tools/moq_chat.h"
#include "quiche/quic/moqt/tools/moqt_client.h"
#include "quiche/common/platform/api/quiche_export.h"
#include "quiche/common/quiche_callbacks.h"
-namespace moqt {
+namespace moqt::moq_chat {
constexpr quic::QuicTime::Delta kChatEventLoopDuration =
quic::QuicTime::Delta::FromMilliseconds(500);
@@ -58,6 +55,8 @@
// an event loop.
ChatClient(const quic::QuicServerId& server_id, bool ignore_certificate,
std::unique_ptr<ChatUserInterface> interface,
+ absl::string_view chat_id, absl::string_view username,
+ absl::string_view localhost,
quic::QuicEventLoop* event_loop = nullptr);
~ChatClient() {
if (session_ != nullptr) {
@@ -67,8 +66,7 @@
}
// Establish the MoQT session. Returns false if it fails.
- bool Connect(absl::string_view path, absl::string_view username,
- absl::string_view chat_id);
+ bool Connect(absl::string_view path);
void OnTerminalLineInput(absl::string_view input_message);
@@ -111,7 +109,7 @@
};
// Returns false on error.
- bool AnnounceAndSubscribe();
+ bool AnnounceAndSubscribeAnnounces();
bool session_is_open() const { return session_is_open_; }
@@ -119,29 +117,18 @@
// catalog, subscribing to all the users in it, and waiting for the server
// to subscribe to the local track.
bool is_syncing() const {
- return !catalog_group_.has_value() || subscribes_to_make_ > 0 ||
+ return subscribes_to_make_ > 0 ||
(queue_ == nullptr || !queue_->HasSubscribers());
}
private:
void RunEventLoop() { event_loop_->RunEventLoopOnce(kChatEventLoopDuration); }
-
- // Objects from the same catalog group arrive on the same stream, and in
- // object sequence order.
- void ProcessCatalog(absl::string_view object,
- moqt::SubscribeRemoteTrack::Visitor* visitor,
- uint64_t group_sequence, uint64_t object_sequence);
-
- struct ChatUser {
- moqt::FullTrackName full_track_name;
- uint64_t from_group;
- ChatUser(const moqt::FullTrackName& ftn, uint64_t group)
- : full_track_name(ftn), from_group(group) {}
- };
+ // Callback for incoming announces.
+ std::optional<MoqtAnnounceErrorReason> OnIncomingAnnounce(
+ const moqt::FullTrackName& track_namespace, AnnounceEvent announce_type);
// Basic session information
- std::string username_;
- std::optional<moqt::MoqChatStrings> chat_strings_;
+ FullTrackName my_track_name_;
// General state variables
// The event loop to use for this client.
@@ -156,13 +143,12 @@
moqt::MoqtSessionCallbacks session_callbacks_;
// Related to syncing.
- std::optional<uint64_t> catalog_group_;
- absl::flat_hash_map<std::string, ChatUser> other_users_;
- int subscribes_to_make_ = 1;
+ absl::flat_hash_set<FullTrackName> other_users_;
+ int subscribes_to_make_ = 0;
// Related to subscriptions/announces
// TODO: One for each subscribe
- std::unique_ptr<RemoteTrackVisitor> remote_track_visitor_;
+ RemoteTrackVisitor remote_track_visitor_;
// Handling outgoing messages
std::shared_ptr<moqt::MoqtOutgoingQueue> queue_;
@@ -171,6 +157,6 @@
std::unique_ptr<ChatUserInterface> interface_;
};
-} // namespace moqt
+} // namespace moqt::moq_chat
#endif // QUICHE_QUIC_MOQT_TOOLS_CHAT_CLIENT_H
diff --git a/quiche/quic/moqt/tools/chat_client_bin.cc b/quiche/quic/moqt/tools/chat_client_bin.cc
index 7617789..5e9125c 100644
--- a/quiche/quic/moqt/tools/chat_client_bin.cc
+++ b/quiche/quic/moqt/tools/chat_client_bin.cc
@@ -31,8 +31,11 @@
std::string, output_file, "",
"chat messages will stream to a file instead of stdout");
+using ::moqt::moq_chat::ChatClient;
+using ::moqt::moq_chat::ChatUserInterface;
+
// Writes messages to a file, when directed from the command line.
-class FileOutput : public moqt::ChatUserInterface {
+class FileOutput : public ChatUserInterface {
public:
explicit FileOutput(absl::string_view filename, absl::string_view username)
: username_(username) {
@@ -66,7 +69,7 @@
event_loop_ == nullptr)
<< "IoLoop called before Initialize";
while (poll(&poll_settings_, 1, 0) <= 0) {
- event_loop_->RunEventLoopOnce(moqt::kChatEventLoopDuration);
+ event_loop_->RunEventLoopOnce(moqt::moq_chat::kChatEventLoopDuration);
}
std::getline(std::cin, message_to_send);
callback_(message_to_send);
@@ -86,7 +89,7 @@
};
// Writes messages to the terminal, without messing up entry of new messages.
-class CliOutput : public moqt::ChatUserInterface {
+class CliOutput : public ChatUserInterface {
public:
void Initialize(quic::InteractiveCli::LineCallback callback,
quic::QuicEventLoop* event_loop) override {
@@ -107,7 +110,7 @@
QUIC_BUG_IF(quic_bug_moq_chat_user_interface_unitialized,
event_loop_ == nullptr)
<< "IoLoop called before Initialize";
- event_loop_->RunEventLoopOnce(moqt::kChatEventLoopDuration);
+ event_loop_->RunEventLoopOnce(moqt::moq_chat::kChatEventLoopDuration);
}
private:
@@ -118,10 +121,11 @@
// A client for MoQT over chat, used for interop testing. See
// https://afrind.github.io/draft-frindell-moq-chat/draft-frindell-moq-chat.html
int main(int argc, char* argv[]) {
- const char* usage = "Usage: chat_client [options] <url> <username> <chat-id>";
+ const char* usage =
+ "Usage: chat_client [options] <url> <username> <chat-id> <localhost>";
std::vector<std::string> args =
quiche::QuicheParseCommandLineFlags(usage, argc, argv);
- if (args.size() != 3) {
+ if (args.size() != 4) {
quiche::QuichePrintCommandLineFlagHelp(usage);
return 1;
}
@@ -130,24 +134,25 @@
std::string path = url.PathParamsQuery();
const std::string& username = args[1];
const std::string& chat_id = args[2];
+ std::string localhost = args[3];
std::string output_filename =
quiche::GetQuicheCommandLineFlag(FLAGS_output_file);
- std::unique_ptr<moqt::ChatUserInterface> interface;
+ std::unique_ptr<ChatUserInterface> interface;
if (!output_filename.empty()) {
interface = std::make_unique<FileOutput>(output_filename, username);
} else { // Use the CLI.
interface = std::make_unique<CliOutput>();
}
- moqt::ChatClient client(
+ ChatClient client(
server_id,
quiche::GetQuicheCommandLineFlag(FLAGS_disable_certificate_verification),
- std::move(interface));
+ std::move(interface), chat_id, username, localhost);
- if (!client.Connect(path, username, chat_id)) {
+ if (!client.Connect(path)) {
return 1;
}
- if (!client.AnnounceAndSubscribe()) {
+ if (!client.AnnounceAndSubscribeAnnounces()) {
return 1;
}
client.IoLoop();
diff --git a/quiche/quic/moqt/tools/chat_server.cc b/quiche/quic/moqt/tools/chat_server.cc
index 243d13b..871390b 100644
--- a/quiche/quic/moqt/tools/chat_server.cc
+++ b/quiche/quic/moqt/tools/chat_server.cc
@@ -10,54 +10,114 @@
#include <string>
#include <utility>
+#include "absl/functional/bind_front.h"
#include "absl/status/status.h"
#include "absl/status/statusor.h"
-#include "absl/strings/str_cat.h"
#include "absl/strings/string_view.h"
#include "quiche/quic/core/crypto/proof_source.h"
#include "quiche/quic/moqt/moqt_live_relay_queue.h"
#include "quiche/quic/moqt/moqt_messages.h"
-#include "quiche/quic/moqt/moqt_outgoing_queue.h"
#include "quiche/quic/moqt/moqt_priority.h"
#include "quiche/quic/moqt/moqt_session.h"
#include "quiche/quic/moqt/tools/moq_chat.h"
#include "quiche/quic/moqt/tools/moqt_server.h"
-#include "quiche/common/platform/api/quiche_mem_slice.h"
-#include "quiche/common/quiche_buffer_allocator.h"
-#include "quiche/common/simple_buffer_allocator.h"
-namespace moqt {
+namespace moqt::moq_chat {
+
+std::optional<MoqtAnnounceErrorReason>
+ChatServer::ChatServerSessionHandler::OnIncomingAnnounce(
+ const moqt::FullTrackName& track_namespace, AnnounceEvent announce_type) {
+ if (track_name_.has_value() &&
+ GetUserNamespace(*track_name_) != track_namespace) {
+ // ChatServer only supports one track per client session at a time. Return
+ // ANNOUNCE_OK and exit.
+ return std::nullopt;
+ }
+ // Accept the ANNOUNCE regardless of the chat_id.
+ track_name_ = ConstructTrackNameFromNamespace(track_namespace,
+ GetChatId(track_namespace));
+ if (!track_name_.has_value()) {
+ std::cout << "Malformed ANNOUNCE namespace\n";
+ return MoqtAnnounceErrorReason(
+ MoqtAnnounceErrorCode::kNotASubscribedNamespace,
+ "Not a valid namespace for this chat.");
+ }
+ if (announce_type == AnnounceEvent::kUnannounce) {
+ std::cout << "Received UNANNOUNCE for " << track_namespace.ToString()
+ << "\n";
+ server_->DeleteUser(*track_name_);
+ track_name_.reset();
+ return std::nullopt;
+ }
+ std::cout << "Received ANNOUNCE for " << track_namespace.ToString() << "\n";
+ session_->SubscribeCurrentGroup(*track_name_,
+ server_->remote_track_visitor());
+ server_->AddUser(*track_name_);
+ return std::nullopt;
+}
+
+void ChatServer::ChatServerSessionHandler::OnOutgoingAnnounceReply(
+ FullTrackName track_namespace,
+ std::optional<MoqtAnnounceErrorReason> error_message) {
+ // Log the result; the server doesn't really care.
+ std::cout << "ANNOUNCE for " << track_namespace.ToString();
+ if (error_message.has_value()) {
+ std::cout << " failed with error: " << error_message->reason_phrase << "\n";
+ } else {
+ std::cout << " succeeded\n";
+ }
+}
ChatServer::ChatServerSessionHandler::ChatServerSessionHandler(
MoqtSession* session, ChatServer* server)
: session_(session), server_(server) {
- session_->callbacks().incoming_announce_callback =
- [&](FullTrackName track_namespace, AnnounceEvent announce_type) {
- FullTrackName track_name = track_namespace;
- track_name.AddElement("");
- if (announce_type == AnnounceEvent::kAnnounce) {
- std::cout << "Received ANNOUNCE for ";
- } else {
- std::cout << "Received UNANNOUNCE for ";
- }
- std::cout << track_namespace.ToString() << "\n";
- username_ = server_->strings().GetUsernameFromFullTrackName(track_name);
- if (username_->empty()) {
- std::cout << "Malformed ANNOUNCE namespace\n";
- return std::nullopt;
- }
- session_->SubscribeCurrentGroup(track_name,
- server_->remote_track_visitor());
- server_->AddUser(*username_);
- return std::nullopt;
- };
- // TODO(martinduke): Add a callback for UNANNOUNCE that deletes the user and
- // clears username_, but keeps the handler.
+ session_->callbacks().incoming_announce_callback = absl::bind_front(
+ &ChatServer::ChatServerSessionHandler::OnIncomingAnnounce, this);
session_->callbacks().session_terminated_callback =
- [&](absl::string_view error_message) {
+ [this](absl::string_view error_message) {
std::cout << "Session terminated, reason = " << error_message << "\n";
session_ = nullptr;
server_->DeleteSession(it_);
+ if (track_name_.has_value()) {
+ server_->DeleteUser(*track_name_);
+ }
+ };
+ session_->callbacks().incoming_subscribe_announces_callback =
+ [this](const moqt::FullTrackName& chat_namespace,
+ SubscribeEvent subscribe_type) {
+ if (subscribe_type == SubscribeEvent::kSubscribe) {
+ subscribed_namespaces_.insert(chat_namespace);
+ std::cout << "Received SUBSCRIBE_ANNOUNCES for ";
+ } else {
+ subscribed_namespaces_.erase(chat_namespace);
+ std::cout << "Received UNSUBSCRIBE_ANNOUNCES for ";
+ }
+ std::cout << chat_namespace.ToString() << "\n";
+ if (!IsValidChatNamespace(chat_namespace)) {
+ std::cout << "Not a valid moq-chat namespace.\n";
+ return std::make_optional(
+ MoqtSubscribeErrorReason{SubscribeErrorCode::kTrackDoesNotExist,
+ "Not a valid namespace for this chat."});
+ }
+ if (subscribe_type == SubscribeEvent::kUnsubscribe) {
+ return std::optional<MoqtSubscribeErrorReason>();
+ }
+ // Send all ANNOUNCE.
+ for (auto& [track_name, queue] : server_->user_queues_) {
+ std::cout << "Sending ANNOUNCE for "
+ << GetUserNamespace(track_name).ToString() << "\n";
+ if (track_name_.has_value() &&
+ GetUsername(*track_name_) == GetUsername(track_name)) {
+ // Don't ANNOUNCE a client to itself.
+ continue;
+ }
+ session_->Announce(
+ GetUserNamespace(track_name),
+ absl::bind_front(&ChatServer::ChatServerSessionHandler::
+ OnOutgoingAnnounceReply,
+ this));
+ }
+ return std::optional<MoqtSubscribeErrorReason>();
};
session_->set_publisher(server_->publisher());
}
@@ -66,8 +126,8 @@
if (!server_->is_running_) {
return;
}
- if (username_.has_value()) {
- server_->DeleteUser(*username_);
+ if (track_name_.has_value()) {
+ server_->DeleteUser(*track_name_);
}
}
@@ -78,20 +138,12 @@
const moqt::FullTrackName& full_track_name,
std::optional<FullSequence> /*largest_id*/,
std::optional<absl::string_view> reason_phrase) {
- std::cout << "Subscription to user "
- << server_->strings().GetUsernameFromFullTrackName(full_track_name)
- << " ";
+ std::cout << "Subscription to " << full_track_name.ToString();
if (reason_phrase.has_value()) {
- std::cout << "REJECTED, reason = " << *reason_phrase << "\n";
- std::string username =
- server_->strings().GetUsernameFromFullTrackName(full_track_name);
- if (!username.empty()) {
- std::cout << "Rejection was for malformed namespace\n";
- return;
- }
- server_->DeleteUser(username);
+ std::cout << " REJECTED, reason = " << *reason_phrase << "\n";
+ server_->DeleteUser(full_track_name);
} else {
- std::cout << "ACCEPTED\n";
+ std::cout << " ACCEPTED\n";
}
}
@@ -103,40 +155,26 @@
std::cerr << "Error: received partial message despite requesting "
"buffering\n";
}
- std::string username =
- server_->strings().GetUsernameFromFullTrackName(full_track_name);
- if (username.empty()) {
- std::cout << "Received user message with malformed namespace\n";
- return;
- }
- auto it = server_->user_queues_.find(username);
+ auto it = server_->user_queues_.find(full_track_name);
if (it == server_->user_queues_.end()) {
- std::cerr << "Error: received message for unknown user " << username
- << "\n";
+ std::cerr << "Error: received message for unknown track "
+ << full_track_name.ToString() << "\n";
return;
}
if (status != MoqtObjectStatus::kNormal) {
it->second->AddObject(sequence, status);
return;
}
- if (!server_->WriteToFile(username, object)) {
- std::cout << username << ": " << object << "\n\n";
+ if (!server_->WriteToFile(GetUsername(full_track_name), object)) {
+ std::cout << GetUsername(full_track_name) << ": " << object << "\n\n";
}
it->second->AddObject(sequence, object);
}
ChatServer::ChatServer(std::unique_ptr<quic::ProofSource> proof_source,
- absl::string_view chat_id, absl::string_view output_file)
+ absl::string_view output_file)
: server_(std::move(proof_source), std::move(incoming_session_callback_)),
- strings_(chat_id),
- catalog_(std::make_shared<MoqtOutgoingQueue>(
- strings_.GetCatalogName(), MoqtForwardingPreference::kSubgroup)),
remote_track_visitor_(this) {
- catalog_->AddObject(quiche::QuicheMemSlice(quiche::QuicheBuffer::Copy(
- quiche::SimpleBufferAllocator::Get(),
- MoqChatStrings::kCatalogHeader)),
- /*key=*/true);
- publisher_.Add(catalog_);
if (!output_file.empty()) {
output_filename_ = output_file;
}
@@ -154,27 +192,30 @@
server_.quic_server().Shutdown();
}
-void ChatServer::AddUser(absl::string_view username) {
- std::string catalog_data = absl::StrCat("+", username);
- catalog_->AddObject(quiche::QuicheMemSlice(quiche::QuicheBuffer::Copy(
- quiche::SimpleBufferAllocator::Get(), catalog_data)),
- /*key=*/false);
+void ChatServer::AddUser(FullTrackName track_name) {
// Add a local track.
- user_queues_[username] = std::make_shared<MoqtLiveRelayQueue>(
- strings_.GetFullTrackNameFromUsername(username),
- MoqtForwardingPreference::kSubgroup);
- publisher_.Add(user_queues_[username]);
+ user_queues_[track_name] = std::make_shared<MoqtLiveRelayQueue>(
+ track_name, MoqtForwardingPreference::kSubgroup);
+ publisher_.Add(user_queues_[track_name]);
+ FullTrackName track_namespace = track_name;
+ track_namespace.NameToNamespace();
+ for (auto& session : sessions_) {
+ session.AnnounceIfSubscribed(track_namespace);
+ }
}
-void ChatServer::DeleteUser(absl::string_view username) {
- // Delete from Catalog.
- std::string catalog_data = absl::StrCat("-", username);
- catalog_->AddObject(quiche::QuicheMemSlice(quiche::QuicheBuffer::Copy(
- quiche::SimpleBufferAllocator::Get(), catalog_data)),
- /*key=*/false);
- user_queues_[username]->RemoveAllSubscriptions();
- user_queues_.erase(username);
- publisher_.Delete(strings_.GetFullTrackNameFromUsername(username));
+void ChatServer::DeleteUser(FullTrackName track_name) {
+ // RemoveAllSubscriptions() sends a SUBSCRIBE_DONE for each.
+ user_queues_[track_name]->RemoveAllSubscriptions();
+ user_queues_.erase(track_name);
+ publisher_.Delete(track_name);
+ FullTrackName track_namespace = GetUserNamespace(track_name);
+ for (auto& session : sessions_) {
+ session.UnannounceIfSubscribed(track_namespace);
+ }
+ if (user_queues_.empty()) {
+ std::cout << "No more users!\n";
+ }
}
bool ChatServer::WriteToFile(absl::string_view username,
@@ -189,7 +230,7 @@
absl::StatusOr<MoqtConfigureSessionCallback> ChatServer::IncomingSessionHandler(
absl::string_view path) {
- if (!strings_.IsValidPath(path)) {
+ if (!IsValidPath(path)) {
return absl::NotFoundError("Unknown endpoint; try \"/moq-chat\".");
}
return [this](MoqtSession* session) {
@@ -199,4 +240,4 @@
};
}
-} // namespace moqt
+} // namespace moqt::moq_chat
diff --git a/quiche/quic/moqt/tools/chat_server.h b/quiche/quic/moqt/tools/chat_server.h
index 0c883ed..5a4c3b9 100644
--- a/quiche/quic/moqt/tools/chat_server.h
+++ b/quiche/quic/moqt/tools/chat_server.h
@@ -5,7 +5,6 @@
#ifndef QUICHE_QUIC_MOQT_TOOLS_CHAT_SERVER_H_
#define QUICHE_QUIC_MOQT_TOOLS_CHAT_SERVER_H_
-#include <cstdint>
#include <fstream>
#include <list>
#include <memory>
@@ -13,26 +12,27 @@
#include <string>
#include "absl/container/flat_hash_map.h"
+#include "absl/container/flat_hash_set.h"
+#include "absl/functional/bind_front.h"
#include "absl/status/statusor.h"
#include "absl/strings/string_view.h"
#include "quiche/quic/core/crypto/proof_source.h"
#include "quiche/quic/moqt/moqt_known_track_publisher.h"
#include "quiche/quic/moqt/moqt_live_relay_queue.h"
#include "quiche/quic/moqt/moqt_messages.h"
-#include "quiche/quic/moqt/moqt_outgoing_queue.h"
#include "quiche/quic/moqt/moqt_priority.h"
#include "quiche/quic/moqt/moqt_publisher.h"
#include "quiche/quic/moqt/moqt_session.h"
#include "quiche/quic/moqt/moqt_track.h"
-#include "quiche/quic/moqt/tools/moq_chat.h"
#include "quiche/quic/moqt/tools/moqt_server.h"
namespace moqt {
+namespace moq_chat {
class ChatServer {
public:
ChatServer(std::unique_ptr<quic::ProofSource> proof_source,
- absl::string_view chat_id, absl::string_view output_file);
+ absl::string_view output_file);
~ChatServer();
class RemoteTrackVisitor : public SubscribeRemoteTrack::Visitor {
@@ -62,12 +62,43 @@
it_ = it;
}
+ void AnnounceIfSubscribed(FullTrackName track_namespace) {
+ for (const FullTrackName& subscribed_namespace : subscribed_namespaces_) {
+ if (track_namespace.InNamespace(subscribed_namespace)) {
+ session_->Announce(
+ track_namespace,
+ absl::bind_front(&ChatServer::ChatServerSessionHandler::
+ OnOutgoingAnnounceReply,
+ this));
+ return;
+ }
+ }
+ }
+
+ void UnannounceIfSubscribed(FullTrackName track_namespace) {
+ for (const FullTrackName& subscribed_namespace : subscribed_namespaces_) {
+ if (track_namespace.InNamespace(subscribed_namespace)) {
+ session_->Unannounce(track_namespace);
+ return;
+ }
+ }
+ }
+
private:
+ // Callback for incoming announces.
+ std::optional<MoqtAnnounceErrorReason> OnIncomingAnnounce(
+ const moqt::FullTrackName& track_namespace,
+ AnnounceEvent announce_type);
+ void OnOutgoingAnnounceReply(
+ FullTrackName track_namespace,
+ std::optional<MoqtAnnounceErrorReason> error_message);
+
MoqtSession* session_; // Not owned.
// This design assumes that each server has exactly one username, although
// in theory there could be multiple users on one session.
- std::optional<std::string> username_;
+ std::optional<FullTrackName> track_name_;
ChatServer* server_; // Not owned.
+ absl::flat_hash_set<FullTrackName> subscribed_namespaces_;
// The iterator of this entry in ChatServer::sessions_, so it can destroy
// itself later.
std::list<ChatServerSessionHandler>::const_iterator it_;
@@ -77,11 +108,9 @@
RemoteTrackVisitor* remote_track_visitor() { return &remote_track_visitor_; }
- MoqtOutgoingQueue* catalog() { return catalog_.get(); }
+ void AddUser(FullTrackName track_name);
- void AddUser(absl::string_view username);
-
- void DeleteUser(absl::string_view username);
+ void DeleteUser(FullTrackName track_name);
void DeleteSession(std::list<ChatServerSessionHandler>::const_iterator it) {
sessions_.erase(it);
@@ -92,8 +121,6 @@
MoqtPublisher* publisher() { return &publisher_; }
- MoqChatStrings& strings() { return strings_; }
-
int num_users() const { return user_queues_.size(); }
private:
@@ -106,16 +133,14 @@
bool is_running_ = true;
MoqtServer server_;
std::list<ChatServerSessionHandler> sessions_;
- MoqChatStrings strings_;
MoqtKnownTrackPublisher publisher_;
- std::shared_ptr<MoqtOutgoingQueue> catalog_;
RemoteTrackVisitor remote_track_visitor_;
- // indexed by username
- absl::flat_hash_map<std::string, std::shared_ptr<MoqtLiveRelayQueue>>
+ absl::flat_hash_map<FullTrackName, std::shared_ptr<MoqtLiveRelayQueue>>
user_queues_;
std::string output_filename_;
std::ofstream output_file_;
};
+} // namespace moq_chat
} // namespace moqt
#endif // QUICHE_QUIC_MOQT_TOOLS_CHAT_SERVER_H_
diff --git a/quiche/quic/moqt/tools/chat_server_bin.cc b/quiche/quic/moqt/tools/chat_server_bin.cc
index 899c258..b905ce3 100644
--- a/quiche/quic/moqt/tools/chat_server_bin.cc
+++ b/quiche/quic/moqt/tools/chat_server_bin.cc
@@ -24,15 +24,16 @@
// A server for MoQT over chat, used for interop testing. See
// https://afrind.github.io/draft-frindell-moq-chat/draft-frindell-moq-chat.html
int main(int argc, char* argv[]) {
- const char* usage = "Usage: chat_server [options] <chat-id>";
+ const char* usage = "Usage: chat_server [options]";
std::vector<std::string> args =
quiche::QuicheParseCommandLineFlags(usage, argc, argv);
- if (args.size() != 1) {
+ if (!args.empty()) {
quiche::QuichePrintCommandLineFlagHelp(usage);
return 1;
}
- moqt::ChatServer server(quiche::CreateDefaultProofSource(), argv[1],
- quiche::GetQuicheCommandLineFlag(FLAGS_output_file));
+ moqt::moq_chat::ChatServer server(
+ quiche::CreateDefaultProofSource(),
+ quiche::GetQuicheCommandLineFlag(FLAGS_output_file));
quiche::QuicheIpAddress bind_address;
QUICHE_CHECK(bind_address.FromString(
quiche::GetQuicheCommandLineFlag(FLAGS_bind_address)));
diff --git a/quiche/quic/moqt/tools/moq_chat.cc b/quiche/quic/moqt/tools/moq_chat.cc
new file mode 100644
index 0000000..0ec5b64
--- /dev/null
+++ b/quiche/quic/moqt/tools/moq_chat.cc
@@ -0,0 +1,76 @@
+// Copyright 2024 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "quiche/quic/moqt/tools/moq_chat.h"
+
+#include <optional>
+
+#include "absl/strings/str_cat.h"
+#include "absl/strings/string_view.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
+#include "quiche/quic/moqt/moqt_messages.h"
+#include "quiche/common/platform/api/quiche_logging.h"
+
+namespace moqt::moq_chat {
+
+bool IsValidPath(absl::string_view path) { return path == kWebtransPath; }
+
+bool IsValidNamespace(const FullTrackName& track_namespace) {
+ return track_namespace.tuple().size() == kFullPathLength - 1 &&
+ track_namespace.tuple()[0] == kBasePath;
+}
+
+bool IsValidChatNamespace(const FullTrackName& track_namespace) {
+ return track_namespace.tuple().size() == 2 &&
+ track_namespace.tuple()[0] == kBasePath;
+}
+
+FullTrackName ConstructTrackName(absl::string_view chat_id,
+ absl::string_view username,
+ absl::string_view device_id) {
+ return FullTrackName{kBasePath,
+ chat_id,
+ username,
+ device_id,
+ absl::StrCat(ToUnixSeconds(::absl::Now())),
+ kNameField};
+}
+
+std::optional<FullTrackName> ConstructTrackNameFromNamespace(
+ const FullTrackName& track_namespace, absl::string_view chat_id) {
+ if (track_namespace.tuple().size() != kFullPathLength - 1) {
+ return std::nullopt;
+ }
+ if (track_namespace.tuple()[0] != kBasePath ||
+ track_namespace.tuple()[1] != chat_id) {
+ return std::nullopt;
+ }
+ FullTrackName track_name = track_namespace;
+ track_name.AddElement(kNameField);
+ return track_name;
+}
+
+absl::string_view GetUsername(const FullTrackName& track_name) {
+ QUICHE_DCHECK(track_name.tuple().size() > 2);
+ return track_name.tuple()[2];
+}
+
+absl::string_view GetChatId(const FullTrackName& track_name) {
+ QUICHE_DCHECK(track_name.tuple().size() > 1);
+ return track_name.tuple()[1];
+}
+
+FullTrackName GetUserNamespace(const FullTrackName& track_name) {
+ QUICHE_DCHECK(track_name.tuple().size() == kFullPathLength);
+ FullTrackName track_namespace = track_name;
+ track_namespace.NameToNamespace();
+ return track_namespace;
+}
+
+FullTrackName GetChatNamespace(const FullTrackName& track_name) {
+ return FullTrackName{track_name.tuple()[0], track_name.tuple()[1]};
+}
+
+} // namespace moqt::moq_chat
diff --git a/quiche/quic/moqt/tools/moq_chat.h b/quiche/quic/moqt/tools/moq_chat.h
index 6789f82..02f23e4 100644
--- a/quiche/quic/moqt/tools/moq_chat.h
+++ b/quiche/quic/moqt/tools/moq_chat.h
@@ -5,64 +5,56 @@
#ifndef QUICHE_QUIC_MOQT_TOOLS_MOQ_CHAT_H
#define QUICHE_QUIC_MOQT_TOOLS_MOQ_CHAT_H
-#include <string>
-#include <vector>
+#include <cstddef>
+#include <optional>
-#include "absl/strings/str_cat.h"
-#include "absl/strings/str_split.h"
#include "absl/strings/string_view.h"
#include "quiche/quic/moqt/moqt_messages.h"
+// Utilities for manipulating moq-chat paths, names, and namespaces.
+
namespace moqt {
-// This class encodes all the syntax in moq-chat strings: paths, full track
-// names, and catalog entries.
-class MoqChatStrings {
- public:
- explicit MoqChatStrings(absl::string_view chat_id) : chat_id_(chat_id) {}
+namespace moq_chat {
- static constexpr absl::string_view kBasePath = "moq-chat";
- static constexpr absl::string_view kParticipantPath = "participant";
- static constexpr absl::string_view kCatalogPath = "catalog";
- static constexpr absl::string_view kCatalogHeader = "version=1\n";
+constexpr absl::string_view kWebtransPath = "/moq-relay";
+// The order of fields is "moq-chat", chat-id, username, device-id, timestamp,
+// "chat".
+// The number of tiers in a full track name.
+constexpr size_t kFullPathLength = 6;
+// The first element in the track namespace or name.
+constexpr absl::string_view kBasePath = "moq-chat";
+// The last element in the track name.
+constexpr absl::string_view kNameField = "chat";
- // Verifies that the WebTransport path matches the spec.
- bool IsValidPath(absl::string_view path) const {
- return path == absl::StrCat("/", kBasePath);
- }
+// Verifies that the WebTransport path matches the spec.
+bool IsValidPath(absl::string_view path);
- // Returns "" if the track namespace is not a participant track.
- std::string GetUsernameFromFullTrackName(
- FullTrackName full_track_name) const {
- if (full_track_name.tuple().size() != 2) {
- return "";
- }
- if (!full_track_name.tuple()[1].empty()) {
- return "";
- }
- std::vector<absl::string_view> elements =
- absl::StrSplit(full_track_name.tuple()[0], '/');
- if (elements.size() != 4 || elements[0] != kBasePath ||
- elements[1] != chat_id_ || elements[2] != kParticipantPath) {
- return "";
- }
- return std::string(elements[3]);
- }
+bool IsValidTrackNamespace(const FullTrackName& track_namespace);
+bool IsValidChatNamespace(const FullTrackName& track_namespace);
- FullTrackName GetFullTrackNameFromUsername(absl::string_view username) const {
- return FullTrackName{absl::StrCat(kBasePath, "/", chat_id_, "/",
- kParticipantPath, "/", username),
- ""};
- }
+// Given a chat-id and username, returns a full track name for moq-chat.
+FullTrackName ConstructTrackName(absl::string_view chat_id,
+ absl::string_view username,
+ absl::string_view device_id);
- FullTrackName GetCatalogName() const {
- return FullTrackName{absl::StrCat(kBasePath, "/", chat_id_),
- absl::StrCat("/", kCatalogPath)};
- }
+// constructs a full track name based on the track_namespace. If the namespace
+// is syntactically incorrect, or does not match the expected value of
+// |chat-id|, returns nullopt
+std::optional<FullTrackName> ConstructTrackNameFromNamespace(
+ const FullTrackName& track_namespace, absl::string_view chat_id);
- private:
- const std::string chat_id_;
-};
+// Strips "chat" from the end of |track_name| to use in ANNOUNCE.
+FullTrackName GetUserNamespace(const FullTrackName& track_name);
+
+// Returns {"moq-chat", chat-id}, useful for SUBSCRIBE_ANNOUNCES.
+FullTrackName GetChatNamespace(const FullTrackName& track_name);
+
+absl::string_view GetUsername(const FullTrackName& track_name);
+
+absl::string_view GetChatId(const FullTrackName& track_name);
+
+} // namespace moq_chat
} // namespace moqt
diff --git a/quiche/quic/moqt/tools/moq_chat_end_to_end_test.cc b/quiche/quic/moqt/tools/moq_chat_end_to_end_test.cc
index 3460aea..8807363 100644
--- a/quiche/quic/moqt/tools/moq_chat_end_to_end_test.cc
+++ b/quiche/quic/moqt/tools/moq_chat_end_to_end_test.cc
@@ -12,6 +12,7 @@
#include "quiche/quic/core/quic_server_id.h"
#include "quiche/quic/moqt/tools/chat_client.h"
#include "quiche/quic/moqt/tools/chat_server.h"
+#include "quiche/quic/moqt/tools/moq_chat.h"
#include "quiche/quic/platform/api/quic_socket_address.h"
#include "quiche/quic/test_tools/crypto_test_utils.h"
#include "quiche/common/platform/api/quiche_test.h"
@@ -26,7 +27,7 @@
constexpr absl::string_view kChatHostname = "127.0.0.1";
-class MockChatUserInterface : public ChatUserInterface {
+class MockChatUserInterface : public moqt::moq_chat::ChatUserInterface {
public:
void Initialize(quiche::MultiUseCallback<void(absl::string_view)> callback,
quic::QuicEventLoop* event_loop) override {
@@ -35,7 +36,7 @@
}
void IoLoop() override {
- event_loop_->RunEventLoopOnce(moqt::kChatEventLoopDuration);
+ event_loop_->RunEventLoopOnce(moqt::moq_chat::kChatEventLoopDuration);
}
MOCK_METHOD(void, WriteToOutput,
@@ -52,8 +53,7 @@
class MoqChatEndToEndTest : public quiche::test::QuicheTest {
public:
MoqChatEndToEndTest()
- : server_(quic::test::crypto_test_utils::ProofSourceForTesting(),
- "test_chat", "") {
+ : server_(quic::test::crypto_test_utils::ProofSourceForTesting(), "") {
quiche::QuicheIpAddress bind_address;
std::string hostname(kChatHostname);
bind_address.FromString(hostname);
@@ -64,11 +64,13 @@
interface1_ = if1ptr.get();
interface2_ = if2ptr.get();
uint16_t port = server_.moqt_server().quic_server().port();
- client1_ = std::make_unique<ChatClient>(
+ client1_ = std::make_unique<moqt::moq_chat::ChatClient>(
quic::QuicServerId(hostname, port), true, std::move(if1ptr),
+ "test_chat", "client1", "device1",
server_.moqt_server().quic_server().event_loop());
- client2_ = std::make_unique<ChatClient>(
+ client2_ = std::make_unique<moqt::moq_chat::ChatClient>(
quic::QuicServerId(hostname, port), true, std::move(if2ptr),
+ "test_chat", "client2", "device2",
server_.moqt_server().quic_server().event_loop());
}
@@ -86,16 +88,16 @@
}
}
- ChatServer server_;
+ moqt::moq_chat::ChatServer server_;
MockChatUserInterface *interface1_, *interface2_;
- std::unique_ptr<ChatClient> client1_, client2_;
+ std::unique_ptr<moqt::moq_chat::ChatClient> client1_, client2_;
};
TEST_F(MoqChatEndToEndTest, EndToEndTest) {
- EXPECT_TRUE(client1_->Connect("/moq-chat", "client1", "test_chat"));
- EXPECT_TRUE(client2_->Connect("/moq-chat", "client2", "test_chat"));
- EXPECT_TRUE(client1_->AnnounceAndSubscribe());
- EXPECT_TRUE(client2_->AnnounceAndSubscribe());
+ EXPECT_TRUE(client1_->Connect(moqt::moq_chat::kWebtransPath));
+ EXPECT_TRUE(client2_->Connect(moqt::moq_chat::kWebtransPath));
+ EXPECT_TRUE(client1_->AnnounceAndSubscribeAnnounces());
+ EXPECT_TRUE(client2_->AnnounceAndSubscribeAnnounces());
SendAndWaitForOutput(interface1_, interface2_, "client1", "Hello");
SendAndWaitForOutput(interface2_, interface1_, "client2", "Hi");
SendAndWaitForOutput(interface1_, interface2_, "client1", "How are you?");
@@ -109,10 +111,10 @@
}
TEST_F(MoqChatEndToEndTest, LeaveAndRejoin) {
- EXPECT_TRUE(client1_->Connect("/moq-chat", "client1", "test_chat"));
- EXPECT_TRUE(client2_->Connect("/moq-chat", "client2", "test_chat"));
- EXPECT_TRUE(client1_->AnnounceAndSubscribe());
- EXPECT_TRUE(client2_->AnnounceAndSubscribe());
+ EXPECT_TRUE(client1_->Connect(moqt::moq_chat::kWebtransPath));
+ EXPECT_TRUE(client2_->Connect(moqt::moq_chat::kWebtransPath));
+ EXPECT_TRUE(client1_->AnnounceAndSubscribeAnnounces());
+ EXPECT_TRUE(client2_->AnnounceAndSubscribeAnnounces());
SendAndWaitForOutput(interface1_, interface2_, "client1", "Hello");
SendAndWaitForOutput(interface2_, interface1_, "client2", "Hi");
@@ -129,11 +131,12 @@
auto if1bptr = std::make_unique<MockChatUserInterface>();
MockChatUserInterface* interface1b_ = if1bptr.get();
uint16_t port = server_.moqt_server().quic_server().port();
- client1_ = std::make_unique<ChatClient>(
+ client1_ = std::make_unique<moqt::moq_chat::ChatClient>(
quic::QuicServerId(std::string(kChatHostname), port), true,
- std::move(if1bptr), server_.moqt_server().quic_server().event_loop());
- EXPECT_TRUE(client1_->Connect("/moq-chat", "client1", "test_chat"));
- EXPECT_TRUE(client1_->AnnounceAndSubscribe());
+ std::move(if1bptr), "test_chat", "client1", "device1",
+ server_.moqt_server().quic_server().event_loop());
+ EXPECT_TRUE(client1_->Connect(moqt::moq_chat::kWebtransPath));
+ EXPECT_TRUE(client1_->AnnounceAndSubscribeAnnounces());
SendAndWaitForOutput(interface1b_, interface2_, "client1", "Hello again");
SendAndWaitForOutput(interface2_, interface1b_, "client2", "Hi again");
}
diff --git a/quiche/quic/moqt/tools/moq_chat_test.cc b/quiche/quic/moqt/tools/moq_chat_test.cc
index f2c8d09..d402054 100644
--- a/quiche/quic/moqt/tools/moq_chat_test.cc
+++ b/quiche/quic/moqt/tools/moq_chat_test.cc
@@ -7,70 +7,63 @@
#include "quiche/quic/moqt/moqt_messages.h"
#include "quiche/common/platform/api/quiche_test.h"
-namespace moqt {
+namespace moqt::moq_chat {
namespace {
-class MoqChatStringsTest : public quiche::test::QuicheTest {
- public:
- MoqChatStrings strings_{"chat-id"};
-};
+class MoqChatTest : public quiche::test::QuicheTest {};
-TEST_F(MoqChatStringsTest, IsValidPath) {
- EXPECT_TRUE(strings_.IsValidPath("/moq-chat"));
- EXPECT_FALSE(strings_.IsValidPath("moq-chat"));
- EXPECT_FALSE(strings_.IsValidPath("/moq-cha"));
- EXPECT_FALSE(strings_.IsValidPath("/moq-chats"));
- EXPECT_FALSE(strings_.IsValidPath("/moq-chat/"));
+TEST_F(MoqChatTest, IsValidPath) {
+ EXPECT_TRUE(IsValidPath("/moq-relay"));
+ EXPECT_FALSE(IsValidPath("moq-relay"));
+ EXPECT_FALSE(IsValidPath("/moq-rela"));
+ EXPECT_FALSE(IsValidPath("/moq-relays"));
+ EXPECT_FALSE(IsValidPath("/moq-relay/"));
}
-TEST_F(MoqChatStringsTest, GetUsernameFromFullTrackName) {
- EXPECT_EQ(strings_.GetUsernameFromFullTrackName(
- FullTrackName{"moq-chat/chat-id/participant/user", ""}),
- "user");
+TEST_F(MoqChatTest, ConstructNameForUser) {
+ FullTrackName name = ConstructTrackName("chat-id", "user", "device");
+
+ EXPECT_EQ(GetChatId(name), "chat-id");
+ EXPECT_EQ(GetUsername(name), "user");
+ // Check that the namespace passes validation.
+ name.NameToNamespace();
+ EXPECT_TRUE(ConstructTrackNameFromNamespace(name, "chat-id").has_value());
}
-TEST_F(MoqChatStringsTest, GetUsernameFromFullTrackNameInvalidInput) {
- EXPECT_EQ(strings_.GetUsernameFromFullTrackName(
- FullTrackName{"/moq-chat/chat-id/participant/user", ""}),
- "");
- EXPECT_EQ(strings_.GetUsernameFromFullTrackName(
- FullTrackName{"moq-chat/chat-id/participant/user/", ""}),
- "");
- EXPECT_EQ(strings_.GetUsernameFromFullTrackName(
- FullTrackName{"moq-cha/chat-id/participant/user", ""}),
- "");
- EXPECT_EQ(strings_.GetUsernameFromFullTrackName(
- FullTrackName{"moq-chat/chat-i/participant/user", ""}),
- "");
- EXPECT_EQ(strings_.GetUsernameFromFullTrackName(
- FullTrackName{"moq-chat/chat-id/participan/user", ""}),
- "");
- EXPECT_EQ(strings_.GetUsernameFromFullTrackName(
- FullTrackName{"moq-chat/chat-id/user", ""}),
- "");
- EXPECT_EQ(strings_.GetUsernameFromFullTrackName(
- FullTrackName{"moq-chat/chat-id/participant/foo/user", ""}),
- "");
- EXPECT_EQ(strings_.GetUsernameFromFullTrackName(
- FullTrackName{"moq-chat/chat-id/participant/user", "foo"}),
- "");
- EXPECT_EQ(strings_.GetUsernameFromFullTrackName(
- FullTrackName{"moq-chat/chat-id/participant/user"}),
- "");
- EXPECT_EQ(strings_.GetUsernameFromFullTrackName(
- FullTrackName{"foo", "moq-chat/chat-id/participant/user", ""}),
- "");
+TEST_F(MoqChatTest, InvalidNamespace) {
+ FullTrackName track_namespace{kBasePath, "chat-id", "username", "device",
+ "timestamp"};
+ // Wrong chat ID.
+ EXPECT_FALSE(
+ ConstructTrackNameFromNamespace(track_namespace, "chat-id2").has_value());
+ // Namespace includes name
+ track_namespace.AddElement("chat");
+ EXPECT_FALSE(
+ ConstructTrackNameFromNamespace(track_namespace, "chat-id").has_value());
+ track_namespace.NameToNamespace(); // Restore to correct value.
+ // Namespace too short.
+ track_namespace.NameToNamespace();
+ EXPECT_FALSE(
+ ConstructTrackNameFromNamespace(track_namespace, "chat-id").has_value());
+ track_namespace.AddElement("chat"); // Restore to correct value.
+ // Base Path is wrong.
+ FullTrackName bad_base_path{"moq-chat2", "chat-id", "user", "device",
+ "timestamp"};
+ EXPECT_FALSE(
+ ConstructTrackNameFromNamespace(bad_base_path, "chat-id").has_value());
}
-TEST_F(MoqChatStringsTest, GetFullTrackNameFromUsername) {
- EXPECT_EQ(strings_.GetFullTrackNameFromUsername("user"),
- FullTrackName("moq-chat/chat-id/participant/user", ""));
-}
-
-TEST_F(MoqChatStringsTest, GetCatalogName) {
- EXPECT_EQ(strings_.GetCatalogName(),
- FullTrackName("moq-chat/chat-id", "/catalog"));
+TEST_F(MoqChatTest, Queries) {
+ FullTrackName local_name{kBasePath, "chat-id", "user",
+ "device", "timestamp", kNameField};
+ EXPECT_EQ(GetChatId(local_name), "chat-id");
+ EXPECT_EQ(GetUsername(local_name), "user");
+ FullTrackName track_namespace{"moq-chat", "chat-id", "user", "device",
+ "timestamp"};
+ EXPECT_EQ(GetUserNamespace(local_name), track_namespace);
+ FullTrackName chat_namespace{"moq-chat", "chat-id"};
+ EXPECT_EQ(GetChatNamespace(local_name), chat_namespace);
}
} // namespace
-} // namespace moqt
+} // namespace moqt::moq_chat
