| // Copyright 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/moqt_session.h" |
| |
| #include <cstdint> |
| #include <memory> |
| #include <optional> |
| #include <string> |
| #include <utility> |
| #include <vector> |
| |
| #include "absl/algorithm/container.h" |
| #include "absl/strings/str_cat.h" |
| #include "absl/strings/string_view.h" |
| #include "quiche/quic/core/quic_types.h" |
| #include "quiche/quic/moqt/moqt_messages.h" |
| #include "quiche/quic/moqt/moqt_subscribe_windows.h" |
| #include "quiche/quic/moqt/moqt_track.h" |
| #include "quiche/common/platform/api/quiche_logging.h" |
| #include "quiche/common/quiche_buffer_allocator.h" |
| #include "quiche/common/quiche_stream.h" |
| #include "quiche/web_transport/web_transport.h" |
| |
| #define ENDPOINT \ |
| (perspective() == Perspective::IS_SERVER ? "MoQT Server: " : "MoQT Client: ") |
| |
| namespace moqt { |
| |
| using ::quic::Perspective; |
| |
| constexpr int kMaxBufferedObjects = 1000; |
| |
| void MoqtSession::OnSessionReady() { |
| QUICHE_DLOG(INFO) << ENDPOINT << "Underlying session ready"; |
| if (parameters_.perspective == Perspective::IS_SERVER) { |
| return; |
| } |
| |
| webtransport::Stream* control_stream = |
| session_->OpenOutgoingBidirectionalStream(); |
| if (control_stream == nullptr) { |
| Error(MoqtError::kGenericError, "Unable to open a control stream"); |
| return; |
| } |
| control_stream->SetVisitor(std::make_unique<Stream>( |
| this, control_stream, /*is_control_stream=*/true)); |
| control_stream_ = control_stream->GetStreamId(); |
| MoqtClientSetup setup = MoqtClientSetup{ |
| .supported_versions = std::vector<MoqtVersion>{parameters_.version}, |
| .role = MoqtRole::kBoth, |
| }; |
| if (!parameters_.using_webtrans) { |
| setup.path = parameters_.path; |
| } |
| quiche::QuicheBuffer serialized_setup = framer_.SerializeClientSetup(setup); |
| bool success = control_stream->Write(serialized_setup.AsStringView()); |
| if (!success) { |
| Error(MoqtError::kGenericError, "Failed to write client SETUP message"); |
| return; |
| } |
| QUIC_DLOG(INFO) << ENDPOINT << "Send the SETUP message"; |
| } |
| |
| void MoqtSession::OnSessionClosed(webtransport::SessionErrorCode, |
| const std::string& error_message) { |
| if (!error_.empty()) { |
| // Avoid erroring out twice. |
| return; |
| } |
| QUICHE_DLOG(INFO) << ENDPOINT << "Underlying session closed with message: " |
| << error_message; |
| error_ = error_message; |
| std::move(session_terminated_callback_)(error_message); |
| } |
| |
| void MoqtSession::OnIncomingBidirectionalStreamAvailable() { |
| while (webtransport::Stream* stream = |
| session_->AcceptIncomingBidirectionalStream()) { |
| if (control_stream_.has_value()) { |
| Error(MoqtError::kProtocolViolation, "Bidirectional stream already open"); |
| return; |
| } |
| stream->SetVisitor(std::make_unique<Stream>(this, stream)); |
| stream->visitor()->OnCanRead(); |
| } |
| } |
| void MoqtSession::OnIncomingUnidirectionalStreamAvailable() { |
| while (webtransport::Stream* stream = |
| session_->AcceptIncomingUnidirectionalStream()) { |
| stream->SetVisitor(std::make_unique<Stream>(this, stream)); |
| stream->visitor()->OnCanRead(); |
| } |
| } |
| |
| void MoqtSession::Error(MoqtError code, absl::string_view error) { |
| if (!error_.empty()) { |
| // Avoid erroring out twice. |
| return; |
| } |
| QUICHE_DLOG(INFO) << ENDPOINT << "MOQT session closed with code: " |
| << static_cast<int>(code) << " and message: " << error; |
| error_ = std::string(error); |
| session_->CloseSession(static_cast<uint64_t>(code), error); |
| std::move(session_terminated_callback_)(error); |
| } |
| |
| void MoqtSession::AddLocalTrack(const FullTrackName& full_track_name, |
| LocalTrack::Visitor* visitor) { |
| local_tracks_.try_emplace(full_track_name, full_track_name, |
| next_track_alias_++, visitor); |
| } |
| |
| // TODO: Create state that allows ANNOUNCE_OK/ERROR on spurious namespaces to |
| // trigger session errors. |
| void MoqtSession::Announce(absl::string_view track_namespace, |
| MoqtAnnounceCallback announce_callback) { |
| if (pending_outgoing_announces_.contains(track_namespace)) { |
| std::move(announce_callback)( |
| track_namespace, "ANNOUNCE message already outstanding for namespace"); |
| return; |
| } |
| MoqtAnnounce message; |
| message.track_namespace = track_namespace; |
| bool success = session_->GetStreamById(*control_stream_) |
| ->Write(framer_.SerializeAnnounce(message).AsStringView()); |
| if (!success) { |
| Error(MoqtError::kGenericError, "Failed to write ANNOUNCE message"); |
| return; |
| } |
| QUIC_DLOG(INFO) << ENDPOINT << "Sent ANNOUNCE message for " |
| << message.track_namespace; |
| pending_outgoing_announces_[track_namespace] = std::move(announce_callback); |
| } |
| |
| bool MoqtSession::HasSubscribers(const FullTrackName& full_track_name) const { |
| auto it = local_tracks_.find(full_track_name); |
| return (it != local_tracks_.end() && it->second.HasSubscriber()); |
| } |
| |
| bool MoqtSession::SubscribeAbsolute(absl::string_view track_namespace, |
| absl::string_view name, |
| uint64_t start_group, uint64_t start_object, |
| RemoteTrack::Visitor* visitor, |
| absl::string_view auth_info) { |
| MoqtSubscribeRequest message; |
| message.track_namespace = track_namespace; |
| message.track_name = name; |
| message.start_group = MoqtSubscribeLocation(true, start_group); |
| message.start_object = MoqtSubscribeLocation(true, start_object); |
| message.end_group = std::nullopt; |
| message.end_object = std::nullopt; |
| if (!auth_info.empty()) { |
| message.authorization_info = std::move(auth_info); |
| } |
| return Subscribe(message, visitor); |
| } |
| |
| bool MoqtSession::SubscribeAbsolute(absl::string_view track_namespace, |
| absl::string_view name, |
| uint64_t start_group, uint64_t start_object, |
| uint64_t end_group, uint64_t end_object, |
| RemoteTrack::Visitor* visitor, |
| absl::string_view auth_info) { |
| if (end_group < start_group) { |
| QUIC_DLOG(ERROR) << "Subscription end is before beginning"; |
| return false; |
| } |
| if (end_group == start_group && end_object < start_object) { |
| QUIC_DLOG(ERROR) << "Subscription end is before beginning"; |
| return false; |
| } |
| MoqtSubscribeRequest message; |
| message.track_namespace = track_namespace; |
| message.track_name = name; |
| message.start_group = MoqtSubscribeLocation(true, start_group); |
| message.start_object = MoqtSubscribeLocation(true, start_object); |
| message.end_group = MoqtSubscribeLocation(true, end_group); |
| message.end_object = MoqtSubscribeLocation(true, end_object); |
| if (!auth_info.empty()) { |
| message.authorization_info = std::move(auth_info); |
| } |
| return Subscribe(message, visitor); |
| } |
| |
| bool MoqtSession::SubscribeRelative(absl::string_view track_namespace, |
| absl::string_view name, int64_t start_group, |
| int64_t start_object, |
| RemoteTrack::Visitor* visitor, |
| absl::string_view auth_info) { |
| MoqtSubscribeRequest message; |
| message.track_namespace = track_namespace; |
| message.track_name = name; |
| message.start_group = MoqtSubscribeLocation(false, start_group); |
| message.start_object = MoqtSubscribeLocation(false, start_object); |
| message.end_group = std::nullopt; |
| message.end_object = std::nullopt; |
| if (!auth_info.empty()) { |
| message.authorization_info = std::move(auth_info); |
| } |
| return Subscribe(message, visitor); |
| } |
| |
| bool MoqtSession::SubscribeCurrentGroup(absl::string_view track_namespace, |
| absl::string_view name, |
| RemoteTrack::Visitor* visitor, |
| absl::string_view auth_info) { |
| MoqtSubscribeRequest message; |
| message.track_namespace = track_namespace; |
| message.track_name = name; |
| // First object of current group. |
| message.start_group = MoqtSubscribeLocation(false, (uint64_t)0); |
| message.start_object = MoqtSubscribeLocation(true, (int64_t)0); |
| message.end_group = std::nullopt; |
| message.end_object = std::nullopt; |
| if (!auth_info.empty()) { |
| message.authorization_info = std::move(auth_info); |
| } |
| return Subscribe(message, visitor); |
| } |
| |
| bool MoqtSession::Subscribe(const MoqtSubscribeRequest& message, |
| RemoteTrack::Visitor* visitor) { |
| // TODO(martinduke): support authorization info |
| bool success = |
| session_->GetStreamById(*control_stream_) |
| ->Write(framer_.SerializeSubscribeRequest(message).AsStringView()); |
| if (!success) { |
| Error(MoqtError::kGenericError, |
| "Failed to write SUBSCRIBE_REQUEST message"); |
| return false; |
| } |
| QUIC_DLOG(INFO) << ENDPOINT << "Sent SUBSCRIBE_REQUEST message for " |
| << message.track_namespace << ":" << message.track_name; |
| FullTrackName ftn(std::string(message.track_namespace), |
| std::string(message.track_name)); |
| remote_tracks_.try_emplace(ftn, ftn, visitor); |
| return true; |
| } |
| |
| std::optional<webtransport::StreamId> MoqtSession::OpenUnidirectionalStream() { |
| if (!session_->CanOpenNextOutgoingUnidirectionalStream()) { |
| return std::nullopt; |
| } |
| webtransport::Stream* new_stream = |
| session_->OpenOutgoingUnidirectionalStream(); |
| if (new_stream == nullptr) { |
| return std::nullopt; |
| } |
| new_stream->SetVisitor(std::make_unique<Stream>(this, new_stream, false)); |
| return new_stream->GetStreamId(); |
| } |
| |
| // increment object_sequence or group_sequence depending on |start_new_group| |
| void MoqtSession::PublishObjectToStream(webtransport::StreamId stream_id, |
| FullTrackName full_track_name, |
| bool start_new_group, |
| absl::string_view payload) { |
| // TODO: check that the peer is subscribed to the next sequence. |
| webtransport::Stream* stream = session_->GetStreamById(stream_id); |
| if (stream == nullptr) { |
| QUICHE_DLOG(ERROR) << ENDPOINT << "Sending OBJECT to nonexistent stream"; |
| return; |
| } |
| auto track_it = local_tracks_.find(full_track_name); |
| if (track_it == local_tracks_.end()) { |
| QUICHE_DLOG(ERROR) << ENDPOINT << "Sending OBJECT for nonexistent track"; |
| return; |
| } |
| MoqtObject object; |
| LocalTrack& track = track_it->second; |
| object.track_id = track.track_alias(); |
| FullSequence& next_sequence = track.next_sequence_mutable(); |
| object.group_sequence = next_sequence.group; |
| if (start_new_group) { |
| ++object.group_sequence; |
| object.object_sequence = 0; |
| } else { |
| object.object_sequence = next_sequence.object; |
| } |
| next_sequence.group = object.group_sequence; |
| next_sequence.object = object.object_sequence + 1; |
| if (!track.ShouldSend(object.group_sequence, object.object_sequence)) { |
| QUICHE_LOG(INFO) << ENDPOINT << "Not sending object " |
| << full_track_name.track_namespace << ":" |
| << full_track_name.track_name << " with sequence " |
| << object.group_sequence << ":" << object.object_sequence |
| << " because peer is not subscribed"; |
| return; |
| } |
| object.object_send_order = 0; |
| object.payload_length = payload.size(); |
| bool success = |
| stream->Write(framer_.SerializeObject(object, payload).AsStringView()); |
| if (!success) { |
| QUICHE_DLOG(ERROR) << ENDPOINT << "Failed to write OBJECT message"; |
| return; |
| } |
| QUICHE_LOG(INFO) << ENDPOINT << "Sending object " |
| << full_track_name.track_namespace << ":" |
| << full_track_name.track_name << " with sequence " |
| << object.group_sequence << ":" << object.object_sequence; |
| } |
| |
| void MoqtSession::Stream::OnCanRead() { |
| bool fin = |
| quiche::ProcessAllReadableRegions(*stream_, [&](absl::string_view chunk) { |
| parser_.ProcessData(chunk, /*end_of_stream=*/false); |
| }); |
| if (fin) { |
| parser_.ProcessData("", /*end_of_stream=*/true); |
| } |
| } |
| void MoqtSession::Stream::OnCanWrite() {} |
| void MoqtSession::Stream::OnResetStreamReceived( |
| webtransport::StreamErrorCode error) { |
| if (is_control_stream_.has_value() && *is_control_stream_) { |
| session_->Error( |
| MoqtError::kProtocolViolation, |
| absl::StrCat("Control stream reset with error code ", error)); |
| } |
| } |
| void MoqtSession::Stream::OnStopSendingReceived( |
| webtransport::StreamErrorCode error) { |
| if (is_control_stream_.has_value() && *is_control_stream_) { |
| session_->Error( |
| MoqtError::kProtocolViolation, |
| absl::StrCat("Control stream reset with error code ", error)); |
| } |
| } |
| |
| void MoqtSession::Stream::OnObjectMessage(const MoqtObject& message, |
| absl::string_view payload, |
| bool end_of_message) { |
| if (is_control_stream_ == true) { |
| session_->Error(MoqtError::kProtocolViolation, |
| "Received OBJECT message on control stream"); |
| return; |
| } |
| QUICHE_DLOG(INFO) << ENDPOINT << "Received OBJECT message on stream " |
| << stream_->GetStreamId() << " for track alias " |
| << message.track_id << " with sequence " |
| << message.group_sequence << ":" << message.object_sequence |
| << " length " << payload.size() << " explicit length " |
| << (message.payload_length.has_value() |
| ? (int)*message.payload_length |
| : -1) |
| << (end_of_message ? "F" : ""); |
| if (!session_->parameters_.deliver_partial_objects) { |
| if (!end_of_message) { // Buffer partial object. |
| absl::StrAppend(&partial_object_, payload); |
| return; |
| } |
| if (!partial_object_.empty()) { // Completes the object |
| absl::StrAppend(&partial_object_, payload); |
| payload = absl::string_view(partial_object_); |
| } |
| } |
| auto it = session_->tracks_by_alias_.find(message.track_id); |
| if (it == session_->tracks_by_alias_.end()) { |
| // No SUBSCRIBE_OK received with this alias, buffer it. |
| auto it2 = session_->object_queue_.find(message.track_id); |
| std::vector<BufferedObject>* queue; |
| if (it2 == session_->object_queue_.end()) { |
| queue = &session_->object_queue_[message.track_id]; |
| } else { |
| queue = &it2->second; |
| } |
| if (session_->num_buffered_objects_ >= kMaxBufferedObjects) { |
| session_->num_buffered_objects_++; |
| session_->Error(MoqtError::kGenericError, "Too many buffered objects"); |
| return; |
| } |
| queue->push_back(BufferedObject(stream_->GetStreamId(), message, payload, |
| end_of_message)); |
| QUIC_DLOG(INFO) << ENDPOINT << "Buffering OBJECT for track alias " |
| << message.track_id; |
| return; |
| } |
| RemoteTrack* subscription = it->second; |
| if (subscription->visitor() != nullptr) { |
| subscription->visitor()->OnObjectFragment( |
| subscription->full_track_name(), stream_->GetStreamId(), |
| message.group_sequence, message.object_sequence, |
| message.object_send_order, payload, end_of_message); |
| } |
| partial_object_.clear(); |
| } |
| |
| void MoqtSession::Stream::OnClientSetupMessage(const MoqtClientSetup& message) { |
| if (is_control_stream_.has_value()) { |
| if (!*is_control_stream_) { |
| session_->Error(MoqtError::kProtocolViolation, |
| "Received SETUP on non-control stream"); |
| return; |
| } |
| } else { |
| is_control_stream_ = true; |
| } |
| if (perspective() == Perspective::IS_CLIENT) { |
| 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(MoqtError::kProtocolViolation, |
| absl::StrCat("Version mismatch: expected 0x", |
| absl::Hex(session_->parameters_.version))); |
| return; |
| } |
| QUICHE_DLOG(INFO) << ENDPOINT << "Received the SETUP message"; |
| if (session_->parameters_.perspective == Perspective::IS_SERVER) { |
| MoqtServerSetup response; |
| response.selected_version = session_->parameters_.version; |
| response.role = MoqtRole::kBoth; |
| bool success = stream_->Write( |
| session_->framer_.SerializeServerSetup(response).AsStringView()); |
| if (!success) { |
| session_->Error(MoqtError::kGenericError, |
| "Failed to write server SETUP message"); |
| return; |
| } |
| QUIC_DLOG(INFO) << ENDPOINT << "Sent the SETUP message"; |
| } |
| // TODO: handle role and path. |
| std::move(session_->session_established_callback_)(); |
| } |
| |
| void MoqtSession::Stream::OnServerSetupMessage(const MoqtServerSetup& message) { |
| if (is_control_stream_.has_value()) { |
| if (!*is_control_stream_) { |
| session_->Error(MoqtError::kProtocolViolation, |
| "Received SETUP on non-control stream"); |
| return; |
| } |
| } else { |
| is_control_stream_ = true; |
| } |
| if (perspective() == Perspective::IS_SERVER) { |
| 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(MoqtError::kProtocolViolation, |
| absl::StrCat("Version mismatch: expected 0x", |
| absl::Hex(session_->parameters_.version))); |
| return; |
| } |
| QUIC_DLOG(INFO) << ENDPOINT << "Received the SETUP message"; |
| // TODO: handle role and path. |
| std::move(session_->session_established_callback_)(); |
| } |
| |
| void MoqtSession::Stream::SendSubscribeError( |
| const MoqtSubscribeRequest& message, SubscribeErrorCode error_code, |
| absl::string_view reason_phrase) { |
| MoqtSubscribeError subscribe_error; |
| subscribe_error.track_namespace = message.track_namespace; |
| subscribe_error.track_name = message.track_name; |
| subscribe_error.error_code = error_code; |
| subscribe_error.reason_phrase = reason_phrase; |
| bool success = |
| stream_->Write(session_->framer_.SerializeSubscribeError(subscribe_error) |
| .AsStringView()); |
| if (!success) { |
| session_->Error(MoqtError::kGenericError, |
| "Failed to write SUBSCRIBE_ERROR message"); |
| } |
| } |
| |
| void MoqtSession::Stream::OnSubscribeRequestMessage( |
| const MoqtSubscribeRequest& message) { |
| std::string reason_phrase = ""; |
| if (!CheckIfIsControlStream()) { |
| return; |
| } |
| QUIC_DLOG(INFO) << ENDPOINT << "Received a SUBSCRIBE_REQUEST for " |
| << message.track_namespace << ":" << message.track_name; |
| auto it = session_->local_tracks_.find(FullTrackName( |
| std::string(message.track_namespace), std::string(message.track_name))); |
| if (it == session_->local_tracks_.end()) { |
| QUIC_DLOG(INFO) << ENDPOINT << "Rejected because " |
| << message.track_namespace << ":" << message.track_name |
| << " does not exist"; |
| SendSubscribeError(message, SubscribeErrorCode::kGenericError, |
| "Track does not exist"); |
| return; |
| } |
| LocalTrack& track = it->second; |
| std::optional<FullSequence> start = session_->LocationToAbsoluteNumber( |
| track, message.start_group, message.start_object); |
| QUICHE_DCHECK(start.has_value()); // Parser enforces this. |
| std::optional<FullSequence> end = session_->LocationToAbsoluteNumber( |
| track, message.end_group, message.end_object); |
| if (start < track.next_sequence() && track.visitor() != nullptr) { |
| SubscribeWindow window = end.has_value() |
| ? SubscribeWindow(start->group, start->object, |
| end->group, end->object) |
| : SubscribeWindow(start->group, start->object); |
| std::optional<absl::string_view> past_objects_available = |
| track.visitor()->OnSubscribeRequestForPast(window); |
| if (!past_objects_available.has_value()) { |
| SendSubscribeError(message, SubscribeErrorCode::kGenericError, |
| "Object does not exist"); |
| return; |
| } |
| } |
| MoqtSubscribeOk subscribe_ok; |
| subscribe_ok.track_namespace = message.track_namespace; |
| subscribe_ok.track_name = message.track_name; |
| subscribe_ok.track_id = track.track_alias(); |
| bool success = stream_->Write( |
| session_->framer_.SerializeSubscribeOk(subscribe_ok).AsStringView()); |
| if (!success) { |
| session_->Error(MoqtError::kGenericError, |
| "Failed to write SUBSCRIBE_OK message"); |
| return; |
| } |
| QUIC_DLOG(INFO) << ENDPOINT << "Created subscription for " |
| << message.track_namespace << ":" << message.track_name; |
| if (!end.has_value()) { |
| track.AddWindow(SubscribeWindow(start->group, start->object)); |
| return; |
| } |
| track.AddWindow( |
| SubscribeWindow(start->group, start->object, end->group, end->object)); |
| } |
| |
| void MoqtSession::Stream::OnSubscribeOkMessage(const MoqtSubscribeOk& message) { |
| if (!CheckIfIsControlStream()) { |
| return; |
| } |
| if (session_->tracks_by_alias_.contains(message.track_id)) { |
| 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(MoqtError::kProtocolViolation, |
| "Received SUBSCRIBE_OK for nonexistent subscribe"); |
| return; |
| } |
| // Note that if there are multiple SUBSCRIBE_OK for the same track, |
| // RemoteTrack.track_alias() will be the last alias received, but |
| // tracks_by_alias_ will have an entry for every track_alias received. |
| // TODO: revise this data structure to make it easier to clean up |
| // RemoteTracks, unless draft changes make it irrelevant. |
| QUIC_DLOG(INFO) << ENDPOINT << "Received the SUBSCRIBE_OK for " |
| << message.track_namespace << ":" << message.track_name |
| << ", track_alias = " << message.track_id; |
| RemoteTrack& track = it->second; |
| track.set_track_alias(message.track_id); |
| session_->tracks_by_alias_[message.track_id] = &track; |
| // TODO: handle expires. |
| if (track.visitor() != nullptr) { |
| track.visitor()->OnReply(track.full_track_name(), std::nullopt); |
| } |
| // Clear the buffer for this track alias. |
| auto it2 = session_->object_queue_.find(message.track_id); |
| if (it2 == session_->object_queue_.end() || track.visitor() == nullptr) { |
| // Nothing is buffered, or the app hasn't registered a visitor anyway. |
| return; |
| } |
| QUIC_DLOG(INFO) << ENDPOINT << "Processing buffered OBJECTs for track_alias " |
| << message.track_id; |
| std::vector<BufferedObject>& queue = it2->second; |
| for (BufferedObject& to_deliver : queue) { |
| track.visitor()->OnObjectFragment( |
| track.full_track_name(), to_deliver.stream_id, |
| to_deliver.message.group_sequence, to_deliver.message.object_sequence, |
| to_deliver.message.object_send_order, to_deliver.payload, |
| to_deliver.eom); |
| session_->num_buffered_objects_--; |
| } |
| session_->object_queue_.erase(it2); |
| } |
| |
| void MoqtSession::Stream::OnSubscribeErrorMessage( |
| const MoqtSubscribeError& message) { |
| if (!CheckIfIsControlStream()) { |
| 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(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 = " << 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); |
| } |
| } |
| |
| void MoqtSession::Stream::OnAnnounceMessage(const MoqtAnnounce& message) { |
| if (!CheckIfIsControlStream()) { |
| return; |
| } |
| MoqtAnnounceOk ok; |
| ok.track_namespace = message.track_namespace; |
| bool success = |
| stream_->Write(session_->framer_.SerializeAnnounceOk(ok).AsStringView()); |
| if (!success) { |
| session_->Error(MoqtError::kGenericError, |
| "Failed to write ANNOUNCE_OK message"); |
| return; |
| } |
| } |
| |
| void MoqtSession::Stream::OnAnnounceOkMessage(const MoqtAnnounceOk& message) { |
| if (!CheckIfIsControlStream()) { |
| return; |
| } |
| auto it = session_->pending_outgoing_announces_.find(message.track_namespace); |
| if (it == session_->pending_outgoing_announces_.end()) { |
| session_->Error(MoqtError::kProtocolViolation, |
| "Received ANNOUNCE_OK for nonexistent announce"); |
| return; |
| } |
| std::move(it->second)(message.track_namespace, std::nullopt); |
| session_->pending_outgoing_announces_.erase(it); |
| } |
| |
| void MoqtSession::Stream::OnAnnounceErrorMessage( |
| const MoqtAnnounceError& message) { |
| if (!CheckIfIsControlStream()) { |
| return; |
| } |
| auto it = session_->pending_outgoing_announces_.find(message.track_namespace); |
| if (it == session_->pending_outgoing_announces_.end()) { |
| session_->Error(MoqtError::kProtocolViolation, |
| "Received ANNOUNCE_ERROR for nonexistent announce"); |
| return; |
| } |
| std::move(it->second)(message.track_namespace, message.reason_phrase); |
| session_->pending_outgoing_announces_.erase(it); |
| } |
| |
| void MoqtSession::Stream::OnParsingError(absl::string_view reason) { |
| session_->Error(MoqtError::kProtocolViolation, |
| absl::StrCat("Parse error: ", reason)); |
| } |
| |
| bool MoqtSession::Stream::CheckIfIsControlStream() { |
| if (!is_control_stream_.has_value()) { |
| session_->Error(MoqtError::kProtocolViolation, |
| "Received SUBSCRIBE_REQUEST as first message"); |
| return false; |
| } |
| if (!*is_control_stream_) { |
| session_->Error(MoqtError::kProtocolViolation, |
| "Received SUBSCRIBE_REQUEST on non-control stream"); |
| return false; |
| } |
| return true; |
| } |
| |
| std::optional<FullSequence> MoqtSession::LocationToAbsoluteNumber( |
| const LocalTrack& track, const std::optional<MoqtSubscribeLocation>& group, |
| const std::optional<MoqtSubscribeLocation>& object) { |
| FullSequence sequence; |
| if (!group.has_value() || !object.has_value()) { |
| return std::nullopt; |
| } |
| if (group->absolute) { |
| sequence.group = group->absolute_value; |
| } else { |
| sequence.group = track.next_sequence().group + group->relative_value; |
| } |
| if (object->absolute) { |
| sequence.object = object->absolute_value; |
| } else { |
| // Subtract 1 because the relative value is computed from the largest sent |
| // sequence number, not the next one. |
| sequence.object = track.next_sequence().object + object->relative_value - 1; |
| } |
| return sequence; |
| } |
| |
| } // namespace moqt |
