Use MOQT standard SubscribeError codes from draft-02.
Also, do what the review for the preceding CL actually meant when it talked about 'enum class'
Move MoqtError in to moqt_messages.h, because it will be needed by the parser.
PiperOrigin-RevId: 601566414
diff --git a/quiche/quic/moqt/moqt_framer.cc b/quiche/quic/moqt/moqt_framer.cc
index a0b5f40..8666061 100644
--- a/quiche/quic/moqt/moqt_framer.cc
+++ b/quiche/quic/moqt/moqt_framer.cc
@@ -288,14 +288,14 @@
NeededVarIntLen(static_cast<uint64_t>(MoqtMessageType::kSubscribeError)) +
LengthPrefixedStringLength(message.track_namespace) +
LengthPrefixedStringLength(message.track_name) +
- NeededVarIntLen(message.error_code) +
+ NeededVarIntLen(static_cast<uint64_t>(message.error_code)) +
LengthPrefixedStringLength(message.reason_phrase);
quiche::QuicheBuffer buffer(allocator_, buffer_size);
quic::QuicDataWriter writer(buffer.size(), buffer.data());
writer.WriteVarInt62(static_cast<uint64_t>(MoqtMessageType::kSubscribeError));
writer.WriteStringPieceVarInt62(message.track_namespace);
writer.WriteStringPieceVarInt62(message.track_name);
- writer.WriteVarInt62(message.error_code);
+ writer.WriteVarInt62(static_cast<uint64_t>(message.error_code));
writer.WriteStringPieceVarInt62(message.reason_phrase);
QUICHE_DCHECK(writer.remaining() == 0);
return buffer;
diff --git a/quiche/quic/moqt/moqt_messages.h b/quiche/quic/moqt/moqt_messages.h
index 773907b..8fd0c55 100644
--- a/quiche/quic/moqt/moqt_messages.h
+++ b/quiche/quic/moqt/moqt_messages.h
@@ -63,6 +63,16 @@
kServerSetup = 0x41,
};
+enum class QUICHE_EXPORT MoqtError : uint64_t {
+ kNoError = 0x0,
+ kGenericError = 0x1,
+ kUnauthorized = 0x2,
+ kProtocolViolation = 0x3,
+ kDuplicateTrackAlias = 0x4,
+ kParameterLengthMismatch = 0x5,
+ kGoawayTimeout = 0x10,
+};
+
enum class QUICHE_EXPORT MoqtRole : uint64_t {
kIngestion = 0x1,
kDelivery = 0x2,
@@ -189,10 +199,16 @@
quic::QuicTimeDelta expires = quic::QuicTimeDelta::FromMilliseconds(0);
};
+enum class QUICHE_EXPORT SubscribeErrorCode : uint64_t {
+ kGenericError = 0x0,
+ kInvalidRange = 0x1,
+ kRetryTrackAlias = 0x2,
+};
+
struct QUICHE_EXPORT MoqtSubscribeError {
absl::string_view track_namespace;
absl::string_view track_name;
- uint64_t error_code;
+ SubscribeErrorCode error_code;
absl::string_view reason_phrase;
};
diff --git a/quiche/quic/moqt/moqt_parser.cc b/quiche/quic/moqt/moqt_parser.cc
index 8a9213c..cde099b 100644
--- a/quiche/quic/moqt/moqt_parser.cc
+++ b/quiche/quic/moqt/moqt_parser.cc
@@ -395,9 +395,11 @@
if (!reader.ReadStringPieceVarInt62(&subscribe_error.track_name)) {
return 0;
}
- if (!reader.ReadVarInt62(&subscribe_error.error_code)) {
+ uint64_t error_code;
+ if (!reader.ReadVarInt62(&error_code)) {
return 0;
}
+ subscribe_error.error_code = static_cast<SubscribeErrorCode>(error_code);
if (!reader.ReadStringPieceVarInt62(&subscribe_error.reason_phrase)) {
return 0;
}
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index adb4689..c79e515 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -30,9 +30,6 @@
using ::quic::Perspective;
-constexpr uint64_t kMoqtErrorTrackDoesntExist = 1;
-constexpr uint64_t kMoqtErrorObjectDoesntExist = 2;
-
constexpr int kMaxBufferedObjects = 1000;
void MoqtSession::OnSessionReady() {
@@ -44,7 +41,7 @@
webtransport::Stream* control_stream =
session_->OpenOutgoingBidirectionalStream();
if (control_stream == nullptr) {
- Error(kGenericError, "Unable to open a control stream");
+ Error(MoqtError::kGenericError, "Unable to open a control stream");
return;
}
control_stream->SetVisitor(std::make_unique<Stream>(
@@ -60,7 +57,7 @@
quiche::QuicheBuffer serialized_setup = framer_.SerializeClientSetup(setup);
bool success = control_stream->Write(serialized_setup.AsStringView());
if (!success) {
- Error(kGenericError, "Failed to write client SETUP message");
+ Error(MoqtError::kGenericError, "Failed to write client SETUP message");
return;
}
QUIC_DLOG(INFO) << ENDPOINT << "Send the SETUP message";
@@ -82,7 +79,7 @@
while (webtransport::Stream* stream =
session_->AcceptIncomingBidirectionalStream()) {
if (control_stream_.has_value()) {
- Error(kProtocolViolation, "Bidirectional stream already open");
+ Error(MoqtError::kProtocolViolation, "Bidirectional stream already open");
return;
}
stream->SetVisitor(std::make_unique<Stream>(this, stream));
@@ -105,7 +102,7 @@
QUICHE_DLOG(INFO) << ENDPOINT << "MOQT session closed with code: "
<< static_cast<int>(code) << " and message: " << error;
error_ = std::string(error);
- session_->CloseSession(code, error);
+ session_->CloseSession(static_cast<uint64_t>(code), error);
std::move(session_terminated_callback_)(error);
}
@@ -129,7 +126,7 @@
bool success = session_->GetStreamById(*control_stream_)
->Write(framer_.SerializeAnnounce(message).AsStringView());
if (!success) {
- Error(kGenericError, "Failed to write ANNOUNCE message");
+ Error(MoqtError::kGenericError, "Failed to write ANNOUNCE message");
return;
}
QUIC_DLOG(INFO) << ENDPOINT << "Sent ANNOUNCE message for "
@@ -230,7 +227,8 @@
session_->GetStreamById(*control_stream_)
->Write(framer_.SerializeSubscribeRequest(message).AsStringView());
if (!success) {
- Error(kGenericError, "Failed to write SUBSCRIBE_REQUEST message");
+ Error(MoqtError::kGenericError,
+ "Failed to write SUBSCRIBE_REQUEST message");
return false;
}
QUIC_DLOG(INFO) << ENDPOINT << "Sent SUBSCRIBE_REQUEST message for "
@@ -319,7 +317,7 @@
webtransport::StreamErrorCode error) {
if (is_control_stream_.has_value() && *is_control_stream_) {
session_->Error(
- kProtocolViolation,
+ MoqtError::kProtocolViolation,
absl::StrCat("Control stream reset with error code ", error));
}
}
@@ -327,7 +325,7 @@
webtransport::StreamErrorCode error) {
if (is_control_stream_.has_value() && *is_control_stream_) {
session_->Error(
- kProtocolViolation,
+ MoqtError::kProtocolViolation,
absl::StrCat("Control stream reset with error code ", error));
}
}
@@ -336,7 +334,7 @@
absl::string_view payload,
bool end_of_message) {
if (is_control_stream_ == true) {
- session_->Error(kProtocolViolation,
+ session_->Error(MoqtError::kProtocolViolation,
"Received OBJECT message on control stream");
return;
}
@@ -371,7 +369,7 @@
}
if (session_->num_buffered_objects_ >= kMaxBufferedObjects) {
session_->num_buffered_objects_++;
- session_->Error(kGenericError, "Too many buffered objects");
+ session_->Error(MoqtError::kGenericError, "Too many buffered objects");
return;
}
queue->push_back(BufferedObject(stream_->GetStreamId(), message, payload,
@@ -393,7 +391,7 @@
void MoqtSession::Stream::OnClientSetupMessage(const MoqtClientSetup& message) {
if (is_control_stream_.has_value()) {
if (!*is_control_stream_) {
- session_->Error(kProtocolViolation,
+ session_->Error(MoqtError::kProtocolViolation,
"Received SETUP on non-control stream");
return;
}
@@ -401,13 +399,14 @@
is_control_stream_ = true;
}
if (perspective() == Perspective::IS_CLIENT) {
- session_->Error(kProtocolViolation, "Received CLIENT_SETUP from server");
+ session_->Error(MoqtError::kProtocolViolation,
+ "Received CLIENT_SETUP from server");
return;
}
if (absl::c_find(message.supported_versions, session_->parameters_.version) ==
message.supported_versions.end()) {
// TODO(martinduke): Is this the right error code? See issue #346.
- session_->Error(kProtocolViolation,
+ session_->Error(MoqtError::kProtocolViolation,
absl::StrCat("Version mismatch: expected 0x",
absl::Hex(session_->parameters_.version)));
return;
@@ -420,7 +419,8 @@
bool success = stream_->Write(
session_->framer_.SerializeServerSetup(response).AsStringView());
if (!success) {
- session_->Error(kGenericError, "Failed to write server SETUP message");
+ session_->Error(MoqtError::kGenericError,
+ "Failed to write server SETUP message");
return;
}
QUIC_DLOG(INFO) << ENDPOINT << "Sent the SETUP message";
@@ -432,7 +432,7 @@
void MoqtSession::Stream::OnServerSetupMessage(const MoqtServerSetup& message) {
if (is_control_stream_.has_value()) {
if (!*is_control_stream_) {
- session_->Error(kProtocolViolation,
+ session_->Error(MoqtError::kProtocolViolation,
"Received SETUP on non-control stream");
return;
}
@@ -440,12 +440,13 @@
is_control_stream_ = true;
}
if (perspective() == Perspective::IS_SERVER) {
- session_->Error(kProtocolViolation, "Received SERVER_SETUP from client");
+ session_->Error(MoqtError::kProtocolViolation,
+ "Received SERVER_SETUP from client");
return;
}
if (message.selected_version != session_->parameters_.version) {
// TODO(martinduke): Is this the right error code? See issue #346.
- session_->Error(kProtocolViolation,
+ session_->Error(MoqtError::kProtocolViolation,
absl::StrCat("Version mismatch: expected 0x",
absl::Hex(session_->parameters_.version)));
return;
@@ -456,7 +457,7 @@
}
void MoqtSession::Stream::SendSubscribeError(
- const MoqtSubscribeRequest& message, uint64_t error_code,
+ const MoqtSubscribeRequest& message, SubscribeErrorCode error_code,
absl::string_view reason_phrase) {
MoqtSubscribeError subscribe_error;
subscribe_error.track_namespace = message.track_namespace;
@@ -467,7 +468,8 @@
stream_->Write(session_->framer_.SerializeSubscribeError(subscribe_error)
.AsStringView());
if (!success) {
- session_->Error(kGenericError, "Failed to write SUBSCRIBE_ERROR message");
+ session_->Error(MoqtError::kGenericError,
+ "Failed to write SUBSCRIBE_ERROR message");
}
}
@@ -485,7 +487,7 @@
QUIC_DLOG(INFO) << ENDPOINT << "Rejected because "
<< message.track_namespace << ":" << message.track_name
<< " does not exist";
- SendSubscribeError(message, kMoqtErrorTrackDoesntExist,
+ SendSubscribeError(message, SubscribeErrorCode::kGenericError,
"Track does not exist");
return;
}
@@ -503,7 +505,7 @@
std::optional<absl::string_view> past_objects_available =
track.visitor()->OnSubscribeRequestForPast(window);
if (!past_objects_available.has_value()) {
- SendSubscribeError(message, kMoqtErrorObjectDoesntExist,
+ SendSubscribeError(message, SubscribeErrorCode::kGenericError,
"Object does not exist");
return;
}
@@ -515,7 +517,8 @@
bool success = stream_->Write(
session_->framer_.SerializeSubscribeOk(subscribe_ok).AsStringView());
if (!success) {
- session_->Error(kGenericError, "Failed to write SUBSCRIBE_OK message");
+ session_->Error(MoqtError::kGenericError,
+ "Failed to write SUBSCRIBE_OK message");
return;
}
QUIC_DLOG(INFO) << ENDPOINT << "Created subscription for "
@@ -533,13 +536,14 @@
return;
}
if (session_->tracks_by_alias_.contains(message.track_id)) {
- session_->Error(kDuplicateTrackAlias, "Received duplicate track_alias");
+ session_->Error(MoqtError::kDuplicateTrackAlias,
+ "Received duplicate track_alias");
return;
}
auto it = session_->remote_tracks_.find(FullTrackName(
std::string(message.track_namespace), std::string(message.track_name)));
if (it == session_->remote_tracks_.end()) {
- session_->Error(kProtocolViolation,
+ session_->Error(MoqtError::kProtocolViolation,
"Received SUBSCRIBE_OK for nonexistent subscribe");
return;
}
@@ -586,13 +590,14 @@
auto it = session_->remote_tracks_.find(FullTrackName(
std::string(message.track_namespace), std::string(message.track_name)));
if (it == session_->remote_tracks_.end()) {
- session_->Error(kProtocolViolation,
+ session_->Error(MoqtError::kProtocolViolation,
"Received SUBSCRIBE_ERROR for nonexistent subscribe");
return;
}
QUIC_DLOG(INFO) << ENDPOINT << "Received the SUBSCRIBE_ERROR for "
<< message.track_namespace << ":" << message.track_name
- << ", error = " << message.reason_phrase;
+ << ", error = " << static_cast<int>(message.error_code)
+ << " (" << message.reason_phrase << ")";
if (it->second.visitor() != nullptr) {
it->second.visitor()->OnReply(it->second.full_track_name(),
message.reason_phrase);
@@ -608,7 +613,8 @@
bool success =
stream_->Write(session_->framer_.SerializeAnnounceOk(ok).AsStringView());
if (!success) {
- session_->Error(kGenericError, "Failed to write ANNOUNCE_OK message");
+ session_->Error(MoqtError::kGenericError,
+ "Failed to write ANNOUNCE_OK message");
return;
}
}
@@ -619,7 +625,7 @@
}
auto it = session_->pending_outgoing_announces_.find(message.track_namespace);
if (it == session_->pending_outgoing_announces_.end()) {
- session_->Error(kProtocolViolation,
+ session_->Error(MoqtError::kProtocolViolation,
"Received ANNOUNCE_OK for nonexistent announce");
return;
}
@@ -634,7 +640,7 @@
}
auto it = session_->pending_outgoing_announces_.find(message.track_namespace);
if (it == session_->pending_outgoing_announces_.end()) {
- session_->Error(kProtocolViolation,
+ session_->Error(MoqtError::kProtocolViolation,
"Received ANNOUNCE_ERROR for nonexistent announce");
return;
}
@@ -643,17 +649,18 @@
}
void MoqtSession::Stream::OnParsingError(absl::string_view reason) {
- session_->Error(kProtocolViolation, absl::StrCat("Parse error: ", reason));
+ session_->Error(MoqtError::kProtocolViolation,
+ absl::StrCat("Parse error: ", reason));
}
bool MoqtSession::Stream::CheckIfIsControlStream() {
if (!is_control_stream_.has_value()) {
- session_->Error(kProtocolViolation,
+ session_->Error(MoqtError::kProtocolViolation,
"Received SUBSCRIBE_REQUEST as first message");
return false;
}
if (!*is_control_stream_) {
- session_->Error(kProtocolViolation,
+ session_->Error(MoqtError::kProtocolViolation,
"Received SUBSCRIBE_REQUEST on non-control stream");
return false;
}
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index ec71181..6620b3c 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h
@@ -47,16 +47,6 @@
MoqtSessionDeletedCallback session_deleted_callback = +[] {};
};
-enum MoqtError : uint64_t {
- kNoError = 0x0,
- kGenericError = 0x1,
- kUnauthorized = 0x2,
- kProtocolViolation = 0x3,
- kDuplicateTrackAlias = 0x4,
- kParameterLengthMismatch = 0x5,
- kGoawayTimeout = 0x10,
-};
-
class QUICHE_EXPORT MoqtSession : public webtransport::SessionVisitor {
public:
MoqtSession(webtransport::Session* session, MoqtSessionParameters parameters,
@@ -182,7 +172,7 @@
private:
friend class test::MoqtSessionPeer;
void SendSubscribeError(const MoqtSubscribeRequest& message,
- uint64_t error_code,
+ SubscribeErrorCode error_code,
absl::string_view reason_phrase);
bool CheckIfIsControlStream();
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc
index 34e7fc3..eb231b2 100644
--- a/quiche/quic/moqt/moqt_session_test.cc
+++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -235,13 +235,16 @@
TEST_F(MoqtSessionTest, Error) {
bool reported_error = false;
- EXPECT_CALL(mock_session_, CloseSession(kParameterLengthMismatch, "foo"))
+ EXPECT_CALL(
+ mock_session_,
+ CloseSession(static_cast<uint64_t>(MoqtError::kParameterLengthMismatch),
+ "foo"))
.Times(1);
EXPECT_CALL(session_callbacks_.session_terminated_callback, Call(_))
.WillOnce([&](absl::string_view error_message) {
reported_error = (error_message == "foo");
});
- session_.Error(kParameterLengthMismatch, "foo");
+ session_.Error(MoqtError::kParameterLengthMismatch, "foo");
EXPECT_TRUE(reported_error);
}
@@ -442,7 +445,7 @@
MoqtSubscribeError error = {
/*track_namespace=*/"foo",
/*track_name=*/"bar",
- /*error_code=*/1,
+ /*error_code=*/SubscribeErrorCode::kInvalidRange,
/*reason_phrase=*/"deadbeef",
};
correct_message = false;
@@ -701,8 +704,9 @@
bool reported_error = false;
EXPECT_CALL(mock_session_, AcceptIncomingBidirectionalStream())
.WillOnce(Return(&mock_stream));
- EXPECT_CALL(mock_session_, CloseSession(kProtocolViolation,
- "Bidirectional stream already open"))
+ EXPECT_CALL(mock_session_,
+ CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation),
+ "Bidirectional stream already open"))
.Times(1);
EXPECT_CALL(session_callbacks_.session_terminated_callback, Call(_))
.WillOnce([&](absl::string_view error_message) {
@@ -745,8 +749,9 @@
bool reported_error = false;
EXPECT_CALL(mock_session_, AcceptIncomingBidirectionalStream())
.WillOnce(Return(&mock_stream));
- EXPECT_CALL(mock_session_, CloseSession(kProtocolViolation,
- "Bidirectional stream already open"))
+ EXPECT_CALL(mock_session_,
+ CloseSession(static_cast<uint64_t>(MoqtError::kProtocolViolation),
+ "Bidirectional stream already open"))
.Times(1);
EXPECT_CALL(session_callbacks_.session_terminated_callback, Call(_))
.WillOnce([&](absl::string_view error_message) {
diff --git a/quiche/quic/moqt/test_tools/moqt_test_message.h b/quiche/quic/moqt/test_tools/moqt_test_message.h
index ada4dea..09ebb64 100644
--- a/quiche/quic/moqt/test_tools/moqt_test_message.h
+++ b/quiche/quic/moqt/test_tools/moqt_test_message.h
@@ -487,7 +487,7 @@
MoqtSubscribeError subscribe_error_ = {
/*track_namespace=*/"foo",
/*track_name=*/"bar",
- /*subscribe=*/1,
+ /*subscribe=*/SubscribeErrorCode::kInvalidRange,
/*reason_phrase=*/"bar",
};
};
diff --git a/quiche/quic/moqt/tools/chat_client_bin.cc b/quiche/quic/moqt/tools/chat_client_bin.cc
index 85772c4..f2af2f7 100644
--- a/quiche/quic/moqt/tools/chat_client_bin.cc
+++ b/quiche/quic/moqt/tools/chat_client_bin.cc
@@ -147,7 +147,8 @@
std::optional<absl::string_view> message) {
if (message.has_value()) {
std::cout << "ANNOUNCE rejected, " << *message << "\n";
- session_->Error(moqt::kGenericError, "Local ANNOUNCE rejected");
+ session_->Error(moqt::MoqtError::kGenericError,
+ "Local ANNOUNCE rejected");
return;
}
std::cout << "ANNOUNCE for " << track_namespace << " accepted\n";
@@ -224,7 +225,7 @@
if (!got_version) {
// Chat server currently does not send version
if (line != "version=1") {
- session_->Error(moqt::kProtocolViolation,
+ session_->Error(moqt::MoqtError::kProtocolViolation,
"Catalog does not begin with version");
return;
}
@@ -278,7 +279,7 @@
subscribes_to_make_++;
} else {
if (it->second.from_group == group_sequence) {
- session_->Error(moqt::kProtocolViolation,
+ session_->Error(moqt::MoqtError::kProtocolViolation,
"User listed twice in Catalog");
return;
}
