Refactor MoQ ChatClient to support end-to-end tests and use moq_chat.h.
PiperOrigin-RevId: 660859814
diff --git a/build/source_list.bzl b/build/source_list.bzl
index 052b6ae..66d3aa4 100644
--- a/build/source_list.bzl
+++ b/build/source_list.bzl
@@ -1521,6 +1521,7 @@
"quic/moqt/moqt_track.h",
"quic/moqt/test_tools/moqt_simulator_harness.h",
"quic/moqt/test_tools/moqt_test_message.h",
+ "quic/moqt/tools/chat_client.h",
"quic/moqt/tools/chat_server.h",
"quic/moqt/tools/moq_chat.h",
"quic/moqt/tools/moqt_client.h",
@@ -1549,6 +1550,7 @@
"quic/moqt/moqt_track.cc",
"quic/moqt/moqt_track_test.cc",
"quic/moqt/test_tools/moqt_simulator_harness.cc",
+ "quic/moqt/tools/chat_client.cc",
"quic/moqt/tools/chat_client_bin.cc",
"quic/moqt/tools/chat_server.cc",
"quic/moqt/tools/chat_server_bin.cc",
diff --git a/build/source_list.gni b/build/source_list.gni
index 04e3a31..ef670d5 100644
--- a/build/source_list.gni
+++ b/build/source_list.gni
@@ -1525,6 +1525,7 @@
"src/quiche/quic/moqt/moqt_track.h",
"src/quiche/quic/moqt/test_tools/moqt_simulator_harness.h",
"src/quiche/quic/moqt/test_tools/moqt_test_message.h",
+ "src/quiche/quic/moqt/tools/chat_client.h",
"src/quiche/quic/moqt/tools/chat_server.h",
"src/quiche/quic/moqt/tools/moq_chat.h",
"src/quiche/quic/moqt/tools/moqt_client.h",
@@ -1553,6 +1554,7 @@
"src/quiche/quic/moqt/moqt_track.cc",
"src/quiche/quic/moqt/moqt_track_test.cc",
"src/quiche/quic/moqt/test_tools/moqt_simulator_harness.cc",
+ "src/quiche/quic/moqt/tools/chat_client.cc",
"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",
diff --git a/build/source_list.json b/build/source_list.json
index a24b008..6feebd1 100644
--- a/build/source_list.json
+++ b/build/source_list.json
@@ -1524,6 +1524,7 @@
"quiche/quic/moqt/moqt_track.h",
"quiche/quic/moqt/test_tools/moqt_simulator_harness.h",
"quiche/quic/moqt/test_tools/moqt_test_message.h",
+ "quiche/quic/moqt/tools/chat_client.h",
"quiche/quic/moqt/tools/chat_server.h",
"quiche/quic/moqt/tools/moq_chat.h",
"quiche/quic/moqt/tools/moqt_client.h",
@@ -1552,6 +1553,7 @@
"quiche/quic/moqt/moqt_track.cc",
"quiche/quic/moqt/moqt_track_test.cc",
"quiche/quic/moqt/test_tools/moqt_simulator_harness.cc",
+ "quiche/quic/moqt/tools/chat_client.cc",
"quiche/quic/moqt/tools/chat_client_bin.cc",
"quiche/quic/moqt/tools/chat_server.cc",
"quiche/quic/moqt/tools/chat_server_bin.cc",
diff --git a/quiche/quic/moqt/moqt_live_relay_queue.h b/quiche/quic/moqt/moqt_live_relay_queue.h
index 41a6270..2dda0db 100644
--- a/quiche/quic/moqt/moqt_live_relay_queue.h
+++ b/quiche/quic/moqt/moqt_live_relay_queue.h
@@ -8,6 +8,7 @@
#include <cstddef>
#include <cstdint>
#include <optional>
+#include <tuple>
#include <utility>
#include <vector>
diff --git a/quiche/quic/moqt/tools/chat_client.cc b/quiche/quic/moqt/tools/chat_client.cc
new file mode 100644
index 0000000..760104b
--- /dev/null
+++ b/quiche/quic/moqt/tools/chat_client.cc
@@ -0,0 +1,277 @@
+// Copyright (c) 2023 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/chat_client.h"
+
+#include <poll.h>
+#include <unistd.h>
+
+#include <cstdint>
+#include <fstream>
+#include <iostream>
+#include <memory>
+#include <optional>
+#include <sstream>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "absl/container/flat_hash_map.h"
+#include "absl/functional/bind_front.h"
+#include "absl/strings/str_cat.h"
+#include "absl/strings/str_split.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"
+#include "quiche/quic/core/io/quic_event_loop.h"
+#include "quiche/quic/core/quic_default_clock.h"
+#include "quiche/quic/core/quic_server_id.h"
+#include "quiche/quic/moqt/moqt_known_track_publisher.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/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"
+#include "quiche/quic/tools/fake_proof_verifier.h"
+#include "quiche/quic/tools/interactive_cli.h"
+#include "quiche/quic/tools/quic_name_lookup.h"
+#include "quiche/common/platform/api/quiche_mem_slice.h"
+#include "quiche/common/quiche_buffer_allocator.h"
+#include "quiche/common/simple_buffer_allocator.h"
+
+moqt::ChatClient::ChatClient(const quic::QuicServerId& server_id,
+ absl::string_view path, absl::string_view username,
+ absl::string_view chat_id, bool ignore_certificate,
+ absl::string_view output_file)
+ : username_(username), chat_strings_(chat_id) {
+ quic::QuicDefaultClock* clock = quic::QuicDefaultClock::Get();
+ std::cout << "Connecting to host " << server_id.host() << " port "
+ << server_id.port() << " path " << path << "\n";
+ event_loop_ = quic::GetDefaultEventLoop()->Create(clock);
+ quic::QuicSocketAddress peer_address =
+ quic::tools::LookupAddress(AF_UNSPEC, server_id);
+ std::unique_ptr<quic::ProofVerifier> verifier;
+ output_filename_ = output_file;
+ if (!output_filename_.empty()) {
+ output_file_.open(output_filename_);
+ output_file_ << "Chat transcript:\n";
+ output_file_.flush();
+ }
+ if (ignore_certificate) {
+ verifier = std::make_unique<quic::FakeProofVerifier>();
+ } else {
+ verifier = quic::CreateDefaultProofVerifier(server_id.host());
+ }
+ client_ = std::make_unique<moqt::MoqtClient>(
+ peer_address, server_id, std::move(verifier), event_loop_.get());
+ session_callbacks_.session_established_callback = [this]() {
+ std::cout << "Session established\n";
+ session_is_open_ = true;
+ if (output_filename_.empty()) { // Use the CLI.
+ cli_ = std::make_unique<quic::InteractiveCli>(
+ event_loop_.get(),
+ absl::bind_front(&ChatClient::OnTerminalLineInput, this));
+ cli_->PrintLine("Fully connected. Enter '/exit' to exit the chat.\n");
+ }
+ };
+ session_callbacks_.session_terminated_callback =
+ [this](absl::string_view error_message) {
+ std::cerr << "Closed session, reason = " << error_message << "\n";
+ session_is_open_ = false;
+ };
+ session_callbacks_.session_deleted_callback = [this]() {
+ session_ = nullptr;
+ };
+ client_->Connect(std::string(path), std::move(session_callbacks_));
+}
+
+void moqt::ChatClient::OnTerminalLineInput(absl::string_view input_message) {
+ if (input_message.empty()) {
+ return;
+ }
+ if (input_message == "/exit") {
+ session_is_open_ = false;
+ return;
+ }
+ quiche::QuicheMemSlice message_slice(quiche::QuicheBuffer::Copy(
+ quiche::SimpleBufferAllocator::Get(), input_message));
+ queue_->AddObject(std::move(message_slice), /*key=*/true);
+}
+
+void moqt::ChatClient::RemoteTrackVisitor::OnReply(
+ const moqt::FullTrackName& full_track_name,
+ 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.track_namespace
+ << " ";
+ }
+ if (reason_phrase.has_value()) {
+ std::cout << "REJECTED, reason = " << *reason_phrase << "\n";
+ } else {
+ std::cout << "ACCEPTED\n";
+ }
+}
+
+void moqt::ChatClient::RemoteTrackVisitor::OnObjectFragment(
+ const moqt::FullTrackName& full_track_name, uint64_t group_sequence,
+ uint64_t object_sequence, moqt::MoqtPriority /*publisher_priority*/,
+ moqt::MoqtObjectStatus /*status*/,
+ moqt::MoqtForwardingPreference /*forwarding_preference*/,
+ 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 (group_sequence < client_->catalog_group_) {
+ std::cout << "Ignoring old catalog";
+ return;
+ }
+ client_->ProcessCatalog(object, this, group_sequence, object_sequence);
+ return;
+ }
+ std::string username = full_track_name.track_namespace;
+ username = username.substr(username.find_last_of('/') + 1);
+ if (!client_->other_users_.contains(username)) {
+ std::cout << "Username " << username << "doesn't exist\n";
+ return;
+ }
+ if (client_->has_output_file()) {
+ client_->WriteToFile(username, object);
+ return;
+ }
+ if (cli_ != nullptr) {
+ std::string full_output = absl::StrCat(username, ": ", object);
+ cli_->PrintLine(full_output);
+ }
+}
+
+bool moqt::ChatClient::AnnounceAndSubscribe() {
+ session_ = client_->session();
+ if (session_ == nullptr) {
+ std::cout << "Failed to connect.\n";
+ return false;
+ }
+ FullTrackName my_track_name =
+ chat_strings_.GetFullTrackNameFromUsername(username_);
+ queue_ = std::make_shared<moqt::MoqtOutgoingQueue>(
+ my_track_name, moqt::MoqtForwardingPreference::kObject);
+ publisher_.Add(queue_);
+ session_->set_publisher(&publisher_);
+ moqt::MoqtOutgoingAnnounceCallback announce_callback =
+ [this](absl::string_view track_namespace,
+ std::optional<moqt::MoqtAnnounceErrorReason> reason) {
+ if (reason.has_value()) {
+ std::cout << "ANNOUNCE rejected, " << reason->reason_phrase << "\n";
+ session_->Error(moqt::MoqtError::kInternalError,
+ "Local ANNOUNCE rejected");
+ return;
+ }
+ std::cout << "ANNOUNCE for " << track_namespace << " accepted\n";
+ return;
+ };
+ std::cout << "Announcing " << my_track_name.track_namespace << "\n";
+ session_->Announce(my_track_name.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.track_namespace, catalog_name.track_name,
+ remote_track_visitor_.get(), username_)) {
+ std::cout << "Failed to get catalog\n";
+ return false;
+ }
+ return true;
+}
+
+void moqt::ChatClient::ProcessCatalog(absl::string_view object,
+ moqt::RemoteTrack::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(moqt::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) {
+ // TODO: Unsubscribe from the user that's leaving
+ 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()) {
+ moqt::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.track_namespace,
+ user_record.full_track_name.track_name, visitor);
+ subscribes_to_make_++;
+ } else {
+ if (it->second.from_group == group_sequence) {
+ session_->Error(moqt::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;
+}
diff --git a/quiche/quic/moqt/tools/chat_client.h b/quiche/quic/moqt/tools/chat_client.h
new file mode 100644
index 0000000..b5cb747
--- /dev/null
+++ b/quiche/quic/moqt/tools/chat_client.h
@@ -0,0 +1,137 @@
+// Copyright (c) 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.
+
+#ifndef QUICHE_QUIC_MOQT_TOOLS_CHAT_CLIENT_H
+#define QUICHE_QUIC_MOQT_TOOLS_CHAT_CLIENT_H
+
+#include <cstdint>
+#include <fstream>
+#include <memory>
+#include <optional>
+#include <string>
+
+#include "absl/container/flat_hash_map.h"
+#include "absl/strings/string_view.h"
+#include "quiche/quic/core/io/quic_event_loop.h"
+#include "quiche/quic/core/quic_server_id.h"
+#include "quiche/quic/core/quic_time.h"
+#include "quiche/quic/moqt/moqt_known_track_publisher.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/moqt_track.h"
+#include "quiche/quic/moqt/tools/moq_chat.h"
+#include "quiche/quic/moqt/tools/moqt_client.h"
+#include "quiche/quic/tools/interactive_cli.h"
+#include "quiche/common/platform/api/quiche_export.h"
+
+namespace moqt {
+
+class ChatClient {
+ public:
+ ChatClient(const quic::QuicServerId& server_id, absl::string_view path,
+ absl::string_view username, absl::string_view chat_id,
+ bool ignore_certificate, absl::string_view output_file);
+
+ void OnTerminalLineInput(absl::string_view input_message);
+
+ bool session_is_open() const { return session_is_open_; }
+
+ // Returns true if the client is still doing initial sync: retrieving the
+ // 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 ||
+ (queue_ == nullptr || !queue_->HasSubscribers());
+ }
+
+ void RunEventLoop() {
+ event_loop_->RunEventLoopOnce(quic::QuicTime::Delta::FromMilliseconds(500));
+ }
+
+ bool has_output_file() { return !output_filename_.empty(); }
+
+ void WriteToFile(absl::string_view user, absl::string_view message) {
+ output_file_ << user << ": " << message << "\n\n";
+ output_file_.flush();
+ }
+
+ class QUICHE_EXPORT RemoteTrackVisitor : public moqt::RemoteTrack::Visitor {
+ public:
+ RemoteTrackVisitor(ChatClient* client) : client_(client) {
+ cli_ = client->cli_.get();
+ }
+
+ void OnReply(const moqt::FullTrackName& full_track_name,
+ std::optional<absl::string_view> reason_phrase) override;
+
+ void OnObjectFragment(const moqt::FullTrackName& full_track_name,
+ uint64_t group_sequence, uint64_t object_sequence,
+ moqt::MoqtPriority publisher_priority,
+ moqt::MoqtObjectStatus status,
+ moqt::MoqtForwardingPreference forwarding_preference,
+ absl::string_view object,
+ bool end_of_message) override;
+
+ void set_cli(quic::InteractiveCli* cli) { cli_ = cli; }
+
+ private:
+ ChatClient* client_;
+ quic::InteractiveCli* cli_;
+ };
+
+ // Returns false on error.
+ bool AnnounceAndSubscribe();
+
+ private:
+ // Objects from the same catalog group arrive on the same stream, and in
+ // object sequence order.
+ void ProcessCatalog(absl::string_view object,
+ moqt::RemoteTrack::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) {}
+ };
+
+ // Basic session information
+ const std::string username_;
+ moqt::MoqChatStrings chat_strings_;
+
+ // General state variables
+ std::unique_ptr<quic::QuicEventLoop> event_loop_;
+ bool session_is_open_ = false;
+ moqt::MoqtSession* session_ = nullptr;
+ moqt::MoqtKnownTrackPublisher publisher_;
+ std::unique_ptr<moqt::MoqtClient> client_;
+ 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;
+
+ // Related to subscriptions/announces
+ // TODO: One for each subscribe
+ std::unique_ptr<RemoteTrackVisitor> remote_track_visitor_;
+
+ // Handling outgoing messages
+ std::shared_ptr<moqt::MoqtOutgoingQueue> queue_;
+
+ // Used when chat output goes to file.
+ std::ofstream output_file_;
+ std::string output_filename_;
+
+ // Used when there is no output file, and both input and output are in the
+ // terminal.
+ std::unique_ptr<quic::InteractiveCli> cli_;
+};
+
+} // namespace moqt
+
+#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 41c0b20..867db4e 100644
--- a/quiche/quic/moqt/tools/chat_client_bin.cc
+++ b/quiche/quic/moqt/tools/chat_client_bin.cc
@@ -5,44 +5,16 @@
#include <poll.h>
#include <unistd.h>
-#include <cstdint>
-#include <fstream>
#include <iostream>
-#include <memory>
-#include <optional>
-#include <sstream>
#include <string>
-#include <utility>
#include <vector>
-#include "absl/container/flat_hash_map.h"
-#include "absl/functional/bind_front.h"
-#include "absl/strings/str_cat.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"
-#include "quiche/quic/core/io/quic_event_loop.h"
-#include "quiche/quic/core/quic_default_clock.h"
#include "quiche/quic/core/quic_server_id.h"
-#include "quiche/quic/core/quic_time.h"
-#include "quiche/quic/moqt/moqt_known_track_publisher.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/moqt_track.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"
-#include "quiche/quic/tools/fake_proof_verifier.h"
-#include "quiche/quic/tools/interactive_cli.h"
+#include "quiche/quic/moqt/tools/chat_client.h"
#include "quiche/quic/tools/quic_name_lookup.h"
#include "quiche/quic/tools/quic_url.h"
#include "quiche/common/platform/api/quiche_command_line_flags.h"
-#include "quiche/common/platform/api/quiche_export.h"
-#include "quiche/common/platform/api/quiche_mem_slice.h"
-#include "quiche/common/quiche_buffer_allocator.h"
-#include "quiche/common/simple_buffer_allocator.h"
DEFINE_QUICHE_COMMAND_LINE_FLAG(
bool, disable_certificate_verification, false,
@@ -52,319 +24,6 @@
std::string, output_file, "",
"chat messages will stream to a file instead of stdout");
-class ChatClient {
- public:
- ChatClient(quic::QuicServerId& server_id, std::string path,
- std::string username, std::string chat_id)
- : chat_id_(chat_id),
- username_(username),
- my_track_name_(UsernameToTrackName(username)),
- catalog_name_("moq-chat/" + chat_id, "/catalog") {
- quic::QuicDefaultClock* clock = quic::QuicDefaultClock::Get();
- std::cout << "Connecting to host " << server_id.host() << " port "
- << server_id.port() << " path " << path << "\n";
- event_loop_ = quic::GetDefaultEventLoop()->Create(clock);
- quic::QuicSocketAddress peer_address =
- quic::tools::LookupAddress(AF_UNSPEC, server_id);
- std::unique_ptr<quic::ProofVerifier> verifier;
- const bool ignore_certificate = quiche::GetQuicheCommandLineFlag(
- FLAGS_disable_certificate_verification);
- output_filename_ = quiche::GetQuicheCommandLineFlag(FLAGS_output_file);
- if (!output_filename_.empty()) {
- output_file_.open(output_filename_);
- output_file_ << "Chat transcript:\n";
- output_file_.flush();
- }
- if (ignore_certificate) {
- verifier = std::make_unique<quic::FakeProofVerifier>();
- } else {
- verifier = quic::CreateDefaultProofVerifier(server_id.host());
- }
- client_ = std::make_unique<moqt::MoqtClient>(
- peer_address, server_id, std::move(verifier), event_loop_.get());
- session_callbacks_.session_established_callback = [this]() {
- std::cout << "Session established\n";
- session_is_open_ = true;
- if (output_filename_.empty()) { // Use the CLI.
- cli_ = std::make_unique<quic::InteractiveCli>(
- event_loop_.get(),
- absl::bind_front(&ChatClient::OnTerminalLineInput, this));
- cli_->PrintLine("Fully connected. Enter '/exit' to exit the chat.\n");
- }
- };
- session_callbacks_.session_terminated_callback =
- [this](absl::string_view error_message) {
- std::cerr << "Closed session, reason = " << error_message << "\n";
- session_is_open_ = false;
- };
- session_callbacks_.session_deleted_callback = [this]() {
- session_ = nullptr;
- };
- client_->Connect(path, std::move(session_callbacks_));
- }
-
- void OnTerminalLineInput(absl::string_view input_message) {
- if (input_message.empty()) {
- return;
- }
- if (input_message == "/exit") {
- session_is_open_ = false;
- return;
- }
- quiche::QuicheMemSlice message_slice(quiche::QuicheBuffer::Copy(
- quiche::SimpleBufferAllocator::Get(), input_message));
- queue_->AddObject(std::move(message_slice), /*key=*/true);
- }
-
- bool session_is_open() const { return session_is_open_; }
- bool is_syncing() const {
- return !catalog_group_.has_value() || subscribes_to_make_ > 0 ||
- (queue_ == nullptr || !queue_->HasSubscribers());
- }
-
- void RunEventLoop() {
- event_loop_->RunEventLoopOnce(quic::QuicTime::Delta::FromMilliseconds(500));
- }
-
- bool has_output_file() { return !output_filename_.empty(); }
-
- void WriteToFile(absl::string_view user, absl::string_view message) {
- output_file_ << user << ": " << message << "\n\n";
- output_file_.flush();
- }
-
- class QUICHE_EXPORT RemoteTrackVisitor : public moqt::RemoteTrack::Visitor {
- public:
- RemoteTrackVisitor(ChatClient* client) : client_(client) {
- cli_ = client->cli_.get();
- }
-
- void OnReply(const moqt::FullTrackName& full_track_name,
- std::optional<absl::string_view> reason_phrase) override {
- client_->subscribes_to_make_--;
- if (full_track_name == client_->catalog_name_) {
- std::cout << "Subscription to catalog ";
- } else {
- std::cout << "Subscription to user " << full_track_name.track_namespace
- << " ";
- }
- if (reason_phrase.has_value()) {
- std::cout << "REJECTED, reason = " << *reason_phrase << "\n";
- } else {
- std::cout << "ACCEPTED\n";
- }
- }
-
- void OnObjectFragment(
- const moqt::FullTrackName& full_track_name, uint64_t group_sequence,
- uint64_t object_sequence, moqt::MoqtPriority /*publisher_priority*/,
- moqt::MoqtObjectStatus /*status*/,
- moqt::MoqtForwardingPreference /*forwarding_preference*/,
- absl::string_view object, bool end_of_message) override {
- if (!end_of_message) {
- std::cerr << "Error: received partial message despite requesting "
- "buffering\n";
- }
- if (full_track_name == client_->catalog_name_) {
- if (group_sequence < client_->catalog_group_) {
- std::cout << "Ignoring old catalog";
- return;
- }
- client_->ProcessCatalog(object, this, group_sequence, object_sequence);
- return;
- }
- std::string username = full_track_name.track_namespace;
- username = username.substr(username.find_last_of('/') + 1);
- if (!client_->other_users_.contains(username)) {
- std::cout << "Username " << username << "doesn't exist\n";
- return;
- }
- if (client_->has_output_file()) {
- client_->WriteToFile(username, object);
- return;
- }
- if (cli_ != nullptr) {
- std::string full_output = absl::StrCat(username, ": ", object);
- cli_->PrintLine(full_output);
- }
- }
-
- void set_cli(quic::InteractiveCli* cli) { cli_ = cli; }
-
- private:
- ChatClient* client_;
- quic::InteractiveCli* cli_;
- };
-
- // returns false on error
- bool AnnounceAndSubscribe() {
- session_ = client_->session();
- if (session_ == nullptr) {
- std::cout << "Failed to connect.\n";
- return false;
- }
- queue_ = std::make_shared<moqt::MoqtOutgoingQueue>(
- my_track_name_, moqt::MoqtForwardingPreference::kObject);
- publisher_.Add(queue_);
- session_->set_publisher(&publisher_);
- moqt::MoqtOutgoingAnnounceCallback announce_callback =
- [&](absl::string_view track_namespace,
- std::optional<moqt::MoqtAnnounceErrorReason> reason) {
- if (reason.has_value()) {
- std::cout << "ANNOUNCE rejected, " << reason->reason_phrase << "\n";
- session_->Error(moqt::MoqtError::kInternalError,
- "Local ANNOUNCE rejected");
- return;
- }
- std::cout << "ANNOUNCE for " << track_namespace << " accepted\n";
- return;
- };
- std::cout << "Announcing " << my_track_name_.track_namespace << "\n";
- session_->Announce(my_track_name_.track_namespace,
- std::move(announce_callback));
- remote_track_visitor_ = std::make_unique<RemoteTrackVisitor>(this);
- if (!session_->SubscribeCurrentGroup(
- catalog_name_.track_namespace, catalog_name_.track_name,
- remote_track_visitor_.get(), username_)) {
- std::cout << "Failed to get catalog for " << chat_id_ << "\n";
- return false;
- }
- return true;
- }
-
- private:
- moqt::FullTrackName UsernameToTrackName(absl::string_view username) {
- return moqt::FullTrackName(
- absl::StrCat("moq-chat/", chat_id_, "/participant/", username), "");
- }
-
- // Objects from the same catalog group arrive on the same stream, and in
- // object sequence order.
- void ProcessCatalog(absl::string_view object,
- moqt::RemoteTrack::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;
- }
- while (std::getline(f, line)) {
- if (!got_version) {
- if (line != "version=1") {
- session_->Error(moqt::MoqtError::kProtocolViolation,
- "Catalog does not begin with version");
- return;
- }
- got_version = true;
- continue;
- }
- if (line.empty()) {
- 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) {
- // TODO: Unsubscribe from the user that's leaving
- 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()) {
- moqt::FullTrackName to_subscribe = UsernameToTrackName(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.track_namespace,
- user_record.full_track_name.track_name, visitor);
- subscribes_to_make_++;
- } else {
- if (it->second.from_group == group_sequence) {
- session_->Error(moqt::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;
- }
-
- struct ChatUser {
- moqt::FullTrackName full_track_name;
- uint64_t from_group;
- ChatUser(moqt::FullTrackName& ftn, uint64_t group)
- : full_track_name(ftn), from_group(group) {}
- };
-
- // Basic session information
- const std::string chat_id_;
- const std::string username_;
- const moqt::FullTrackName my_track_name_;
-
- // General state variables
- std::unique_ptr<quic::QuicEventLoop> event_loop_;
- bool session_is_open_ = false;
- moqt::MoqtSession* session_ = nullptr;
- moqt::MoqtKnownTrackPublisher publisher_;
- std::unique_ptr<moqt::MoqtClient> client_;
- moqt::MoqtSessionCallbacks session_callbacks_;
-
- // Related to syncing.
- std::optional<uint64_t> catalog_group_;
- moqt::FullTrackName catalog_name_;
- absl::flat_hash_map<std::string, ChatUser> other_users_;
- int subscribes_to_make_ = 1;
-
- // Related to subscriptions/announces
- // TODO: One for each subscribe
- std::unique_ptr<RemoteTrackVisitor> remote_track_visitor_;
-
- // Handling incoming and outgoing messages
- std::shared_ptr<moqt::MoqtOutgoingQueue> queue_;
-
- // Used when chat output goes to file.
- std::ofstream output_file_;
- std::string output_filename_;
-
- // Used when there is no output file, and both input and output are in the
- // terminal.
- std::unique_ptr<quic::InteractiveCli> cli_;
-};
-
// 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[]) {
@@ -380,7 +39,10 @@
std::string path = url.PathParamsQuery();
const std::string& username = args[1];
const std::string& chat_id = args[2];
- ChatClient client(server_id, path, username, chat_id);
+ moqt::ChatClient client(
+ server_id, path, username, chat_id,
+ quiche::GetQuicheCommandLineFlag(FLAGS_disable_certificate_verification),
+ quiche::GetQuicheCommandLineFlag(FLAGS_output_file));
while (!client.session_is_open()) {
client.RunEventLoop();
diff --git a/quiche/quic/moqt/tools/moq_chat.h b/quiche/quic/moqt/tools/moq_chat.h
index 37e5278..e2f9041 100644
--- a/quiche/quic/moqt/tools/moq_chat.h
+++ b/quiche/quic/moqt/tools/moq_chat.h
@@ -8,7 +8,6 @@
#include <string>
#include <vector>
-#include "absl/strings/match.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/str_split.h"
#include "absl/strings/string_view.h"
