【WebRTC】视频发送链路中类的简单分析(下)

目录

  • 1.任务队列节流发送器(TaskQueuePacedSender)
    • 1.1 节流控制器添加RTP数据包(PacingController::EnqueuePacket())
    • 1.2 监测是否要处理Packet(PacingController::MaybeProcessPackets())
  • 2.数据包路由(PacketRouter)
  • 3.模块RtpRtcp的实现(ModuleRtpRtcpImpl2)
  • 4.RTP出站发送器(RTPSenderEgress)
  • 5.媒体通道的实现(MediaChannelUtil)
  • 6.基础通道(BaseChannel)
  • 6.安全RTP传输(SrtpTransport)
  • 7.DTLS-SRTP 传输(DtlsSrtpTransport)
  • 8.DTLS传输(DtlsTransport)
  • 9.P2P传输通道(P2PTransportChannel)

WebRTC中类的简单分析:
【WebRTC】视频发送链路中类的简单分析(上)

在前一部分当中,记录视频流已经传输到RTPSender当中的paced_sender_->EnqueuePackets(),这个函数会逐渐向底层深入,发送RTP数据包,下面记录后一部分的传输流程

1.任务队列节流发送器(TaskQueuePacedSender)

paced_sender_->EnqueuePackets()的具体实现位于TaskQueuePacedSender这个类中,声明在modules/pacing/task_queue_paced_sender.h,其中会对任务队列进行一些处理,例如暂停队列,重启队列,检查队列信息等,最核心的函数是EnqueuePackets(),用于将RTP数据包添加到队列当中

class TaskQueuePacedSender : public RtpPacketPacer, public RtpPacketSender {public:static const int kNoPacketHoldback;// The pacer can be configured using `field_trials` or specified parameters.//// The `hold_back_window` parameter sets a lower bound on time to sleep if// there is currently a pacer queue and packets can't immediately be// processed. Increasing this reduces thread wakeups at the expense of higher// latency.//// The taskqueue used when constructing a TaskQueuePacedSender will also be// used for pacing./*节流器可以通过 `field_trials` 进行配置,或者指定参数。`hold_back_window` 参数设置了如果当前有节流队列且数据包不能立即被处理时,睡眠时间的下限。增加这个值可以减少线程唤醒次数,但代价是增加延迟。在构造 TaskQueuePacedSender 时使用的 taskqueue 也将用于节流。*/TaskQueuePacedSender(Clock* clock,PacingController::PacketSender* packet_sender,const FieldTrialsView& field_trials,TimeDelta max_hold_back_window,int max_hold_back_window_in_packets);~TaskQueuePacedSender() override;// The pacer is allowed to send enqued packets in bursts and can build up a// packet "debt" that correspond to approximately the send rate during// 'burst_interval'.// 节流器被允许以突发的方式发送排队的数据包,并可以累积一个“债务”数据包数量,// 这个数量大致对应于在 'burst_interval' 期间的发送速率。// 设置发送突发间隔void SetSendBurstInterval(TimeDelta burst_interval);// A probe may be sent without first waing for a media packet.// 一个探测包可能会在没有首先等待媒体数据包的情况下被发送void SetAllowProbeWithoutMediaPacket(bool allow);// Ensure that necessary delayed tasks are scheduled.// 确保必要的延迟任务被安排void EnsureStarted();// Methods implementing RtpPacketSender.// Adds the packet to the queue and calls// PacingController::PacketSender::SendPacket() when it's time to send.// 将数据包添加到队列当中void EnqueuePackets(std::vector<std::unique_ptr<RtpPacketToSend>> packets) override;// Remove any pending packets matching this SSRC from the packet queue.// 从数据包队列中移除所有与此SSRC匹配的待处理数据包void RemovePacketsForSsrc(uint32_t ssrc) override;// Methods implementing RtpPacketPacer.// 创建探测集群(cluster),网络拥塞控制中,探测集群是一种用来估计网络带宽的方法void CreateProbeClusters(std::vector<ProbeClusterConfig> probe_cluster_configs) override;// Temporarily pause all sending.// 暂停发送数据包void Pause() override;// Resume sending packets.// 重新发送数据包void Resume() override;// 设置网络是否处于拥塞状态void SetCongested(bool congested) override;// Sets the pacing rates. Must be called once before packets can be sent.// 设置发送RTP包的节流速率和填充速率void SetPacingRates(DataRate pacing_rate, DataRate padding_rate) override;// Currently audio traffic is not accounted for by pacer and passed through.// With the introduction of audio BWE, audio traffic will be accounted for// in the pacer budget calculation. The audio traffic will still be injected// at high priority.// 指示节流器是否应该考虑音频流量void SetAccountForAudioPackets(bool account_for_audio) override;// 是否应该在计算发送速率时考虑传输开销。如果启用,发送器将在计算节流速率时包括传输开销,如IP和UDP头部等void SetIncludeOverhead() override;// 设置每个传输包的开销大小。overhead_per_packet参数指定了每个RTP包的固定开销大小,包括IP、UDP和可能的其他协议头部void SetTransportOverhead(DataSize overhead_per_packet) override;// Returns the time since the oldest queued packet was enqueued.// 返回队列中最老的包自入队以来的等待时间TimeDelta OldestPacketWaitTime() const override;// Returns total size of all packets in the pacer queue.// 返回节流器队列中所有包的总大小DataSize QueueSizeData() const override;// Returns the time when the first packet was sent;// 返回发送的第一个数据包的时间戳std::optional<Timestamp> FirstSentPacketTime() const override;// Returns the number of milliseconds it will take to send the current// packets in the queue, given the current size and bitrate, ignoring prio.// 根据当前队列中数据包的大小和比特率,忽略优先级,返回发送当前队列中数据包所需的毫秒数TimeDelta ExpectedQueueTime() const override;// Set the max desired queuing delay, pacer will override the pacing rate// specified by SetPacingRates() if needed to achieve this goal.// 设置最大期望的排队延迟,如果需要的话,节流器将覆盖通过 SetPacingRates() 指定的节流速率以实现这一目标void SetQueueTimeLimit(TimeDelta limit) override;protected:// Exposed as protected for test.struct Stats {Stats(): oldest_packet_enqueue_time(Timestamp::MinusInfinity()),queue_size(DataSize::Zero()),expected_queue_time(TimeDelta::Zero()) {}Timestamp oldest_packet_enqueue_time;DataSize queue_size;TimeDelta expected_queue_time;std::optional<Timestamp> first_sent_packet_time;};void OnStatsUpdated(const Stats& stats);private:// Call in response to state updates that could warrant sending out packets.// Protected against re-entry from packet sent receipts.// 检查当前是否是发送数据包的合适时机,如果是,则直接调用ProcessPackets()来发送数据包;// 如果不是,则安排一个延迟任务,在将来的某个时间点再次检查// 决定是否需要根据状态更新来发送数据包。它通常在网络状态变化、新数据包到达或其他可能影响发送计划的事件发生时被调用void MaybeScheduleProcessPackets() RTC_RUN_ON(task_queue_);// Check if it is time to send packets, or schedule a delayed task if not.// Use Timestamp::MinusInfinity() to indicate that this call has _not_// been scheduled by the pacing controller. If this is the case, check if we// can execute immediately otherwise schedule a delay task that calls this// method again with desired (finite) scheduled process time.// 检查是否是一个合适的时机去发送数据包/*1.如果scheduled_process_time是Timestamp::MinusInfinity(),这表示这个调用没有被节流控制器安排,而是由于其他原因(如新数据包到达)被触发。在这种情况下,函数会检查是否可以立即执行ProcessPackets();如果不可以,则安排一个延迟任务,在合适的时间再次调用MaybeProcessPackets(),并传入一个具体的计划发送时间2.如果scheduled_process_time是一个具体的有限值,函数会检查当前时间是否已经到达或超过了这个计划时间。如果是,它会调用ProcessPackets()来发送数据包;如果不是,它会安排一个延迟任务,在计划时间到达时再次调用MaybeProcessPackets()*/void MaybeProcessPackets(Timestamp scheduled_process_time);void UpdateStats() RTC_RUN_ON(task_queue_);Stats GetStats() const;Clock* const clock_;// The holdback window prevents too frequent delayed MaybeProcessPackets()// calls. These are only applicable if `allow_low_precision` is false.const TimeDelta max_hold_back_window_;const int max_hold_back_window_in_packets_;// 节流控制器PacingController pacing_controller_ RTC_GUARDED_BY(task_queue_);// We want only one (valid) delayed process task in flight at a time.// If the value of `next_process_time_` is finite, it is an id for a// delayed task that will call MaybeProcessPackets() with that time// as parameter.// Timestamp::MinusInfinity() indicates no valid pending task./*我们希望在任何时候只有一个(有效的)延迟处理任务在执行中。如果 `next_process_time_` 的值是有限的,那么它就是一个延迟任务的标识符,这个任务将会使用那个时间作为参数调用 `MaybeProcessPackets()`。Timestamp::MinusInfinity() 表示没有有效的待处理任务*/Timestamp next_process_time_ RTC_GUARDED_BY(task_queue_);// Indicates if this task queue is started. If not, don't allow// posting delayed tasks yet.// 如果当前任务队列没有开始,不允许延时任务bool is_started_ RTC_GUARDED_BY(task_queue_);// Indicates if this task queue is shutting down. If so, don't allow// posting any more delayed tasks as that can cause the task queue to// never drain.// 如果任务队列关闭,不允许添加延时任务bool is_shutdown_ RTC_GUARDED_BY(task_queue_);// Filtered size of enqueued packets, in bytes.rtc::ExpFilter packet_size_ RTC_GUARDED_BY(task_queue_);bool include_overhead_ RTC_GUARDED_BY(task_queue_);Stats current_stats_ RTC_GUARDED_BY(task_queue_);// Protects against ProcessPackets reentry from packet sent receipts.bool processing_packets_ RTC_GUARDED_BY(task_queue_) = false;ScopedTaskSafety safety_;TaskQueueBase* task_queue_;
};

TaskQueuePacedSender::EnqueuePackets()的实现方式如下

void TaskQueuePacedSender::EnqueuePackets(std::vector<std::unique_ptr<RtpPacketToSend>> packets) {task_queue_->PostTask(SafeTask(safety_.flag(), [this, packets = std::move(packets)]() mutable {RTC_DCHECK_RUN_ON(task_queue_);TRACE_EVENT0(TRACE_DISABLED_BY_DEFAULT("webrtc"),"TaskQueuePacedSender::EnqueuePackets");for (auto& packet : packets) {TRACE_EVENT2(TRACE_DISABLED_BY_DEFAULT("webrtc"),"TaskQueuePacedSender::EnqueuePackets::Loop","sequence_number", packet->SequenceNumber(),"rtp_timestamp", packet->Timestamp());size_t packet_size = packet->payload_size() + packet->padding_size();if (include_overhead_) {packet_size += packet->headers_size();}packet_size_.Apply(1, packet_size);RTC_DCHECK_GE(packet->capture_time(), Timestamp::Zero());// 将RTP数据包添加到PacingController当中pacing_controller_.EnqueuePacket(std::move(packet));}// 检查当前时刻是否可以发送数据包MaybeProcessPackets(Timestamp::MinusInfinity());}));
}

上面核心的处理函数为pacing_controller_.EnqueuePacket()和MaybeProcessPackets(),功能分别是添加RTP数据包到PacingController中和检查是否可以发送数据包。

1.1 节流控制器添加RTP数据包(PacingController::EnqueuePacket())

void PacingController::EnqueuePacket(std::unique_ptr<RtpPacketToSend> packet) {// ...// 探测器检测到packetprober_.OnIncomingPacket(DataSize::Bytes(packet->payload_size()));const Timestamp now = CurrentTime();if (packet_queue_.Empty()) {// If queue is empty, we need to "fast-forward" the last process time,// so that we don't use passed time as budget for sending the first new// packet.Timestamp target_process_time = now;Timestamp next_send_time = NextSendTime();if (next_send_time.IsFinite()) {// There was already a valid planned send time, such as a keep-alive.// Use that as last process time only if it's prior to now.target_process_time = std::min(now, next_send_time);}UpdateBudgetWithElapsedTime(UpdateTimeAndGetElapsed(target_process_time));}// 将packet送入到队列当中packet_queue_.Push(now, std::move(packet));seen_first_packet_ = true;// Queue length has increased, check if we need to change the pacing rate.MaybeUpdateMediaRateDueToLongQueue(now);
}

1.2 监测是否要处理Packet(PacingController::MaybeProcessPackets())

函数的作用是检查是否要处理当前队列中的packets,其中最核心的函数是ProcessPackets(),表示处理packets

void TaskQueuePacedSender::MaybeProcessPackets(Timestamp scheduled_process_time) {// ...// Process packets and update stats.while (next_send_time <= now + early_execute_margin) {// 处理packetspacing_controller_.ProcessPackets();next_send_time = pacing_controller_.NextSendTime();RTC_DCHECK(next_send_time.IsFinite());// Probing state could change. Get margin after process packets.early_execute_margin = pacing_controller_.IsProbing()? PacingController::kMaxEarlyProbeProcessing: TimeDelta::Zero();}UpdateStats();// ...
}

pacing_controller_.ProcessPackets()的实现方式如下,主要使用packet_sender_->SendPacket()来发送数据

void PacingController::ProcessPackets() {// ...if (ShouldSendKeepalive(now)) {DataSize keepalive_data_sent = DataSize::Zero();// We can not send padding unless a normal packet has first been sent. If// we do, timestamps get messed up.if (seen_first_packet_) {std::vector<std::unique_ptr<RtpPacketToSend>> keepalive_packets =packet_sender_->GeneratePadding(DataSize::Bytes(1));for (auto& packet : keepalive_packets) {keepalive_data_sent +=DataSize::Bytes(packet->payload_size() + packet->padding_size());packet_sender_->SendPacket(std::move(packet), PacedPacketInfo());for (auto& packet : packet_sender_->FetchFec()) {EnqueuePacket(std::move(packet));}}}OnPacketSent(RtpPacketMediaType::kPadding, keepalive_data_sent, now);}// ...DataSize data_sent = DataSize::Zero();int iteration = 0;int packets_sent = 0;int padding_packets_generated = 0;for (; iteration < circuit_breaker_threshold_; ++iteration) {// ...if (rtp_packet == nullptr) {// ...} else {// ...packet_sender_->SendPacket(std::move(rtp_packet), pacing_info);for (auto& packet : packet_sender_->FetchFec()) {EnqueuePacket(std::move(packet));}// ...}}
}

这里的packet_sender_->SendPacket()的实现由PacketRouter给出

2.数据包路由(PacketRouter)

PacketRouter的声明位于modules/pacing/packet_router.h中,这是一个路由器的类,主要的功能是接收RTP数据包,将其送入到下一模块中。另外,PacketRouter能够发送一些反馈消息

// PacketRouter keeps track of rtp send modules to support the pacer.
// In addition, it handles feedback messages, which are sent on a send
// module if possible (sender report), otherwise on receive module
// (receiver report). For the latter case, we also keep track of the
// receive modules.
/*PacketRouter 负责跟踪 RTP 发送模块,以支持节奏发送器(pacer)。此外,它还处理反馈消息,这些消息如果可能会在发送模块上发送(发送报告),否则则在接收模块上发送(接收报告)。对于后者情况,我们还会跟踪接收模块。
*/
class PacketRouter : public PacingController::PacketSender {public:PacketRouter();~PacketRouter() override;PacketRouter(const PacketRouter&) = delete;PacketRouter& operator=(const PacketRouter&) = delete;// Callback is invoked after pacing, before a packet is forwarded to the// sending rtp module.// 添加一些回调函数,可以在发送数据包之前做一些定义,例如监控带宽等void RegisterNotifyBweCallback(absl::AnyInvocable<void(const RtpPacketToSend& packet,const PacedPacketInfo& pacing_info)> callback);// 添加一个Rtp发送模块(PacketRouter能够管理多个Rtp模块),REMB:Receiver Estimated Max Bitratevoid AddSendRtpModule(RtpRtcpInterface* rtp_module, bool remb_candidate);// 移除一个Rtp发送模块void RemoveSendRtpModule(RtpRtcpInterface* rtp_module);// 检查当前的 PacketRouter 是否支持RTX 载填充,如果支持则具备更好的鲁棒性bool SupportsRtxPayloadPadding() const;// 添加一个Rtp接收模块void AddReceiveRtpModule(RtcpFeedbackSenderInterface* rtcp_sender,bool remb_candidate);// 移除一个Rtp接收模块void RemoveReceiveRtpModule(RtcpFeedbackSenderInterface* rtcp_sender);// 核心函数,发送数据包到下一模块void SendPacket(std::unique_ptr<RtpPacketToSend> packet,const PacedPacketInfo& cluster_info) override;// 获取前向纠错std::vector<std::unique_ptr<RtpPacketToSend>> FetchFec() override;std::vector<std::unique_ptr<RtpPacketToSend>> GeneratePadding(DataSize size) override;// 终止重传请求void OnAbortedRetransmissions(uint32_t ssrc,rtc::ArrayView<const uint16_t> sequence_numbers) override;// 查询与给定媒体 SSRC 相关联的 RTX(重传扩展)SSRCstd::optional<uint32_t> GetRtxSsrcForMedia(uint32_t ssrc) const override;// 在一批数据包处理完成后被调用void OnBatchComplete() override;// Send REMB feedback.// 发送 REMB 反馈void SendRemb(int64_t bitrate_bps, std::vector<uint32_t> ssrcs);// Sends `packets` in one or more IP packets.// 将传入的 RTCP 数据包发送出去。这些数据包可能会被组合成一个或多个 IP 数据包进行发送void SendCombinedRtcpPacket(std::vector<std::unique_ptr<rtcp::RtcpPacket>> packets);private:// 添加一个候选的 REMB 模块void AddRembModuleCandidate(RtcpFeedbackSenderInterface* candidate_module,bool media_sender);// 条件性地移除一个候选的 REMB 模块void MaybeRemoveRembModuleCandidate(RtcpFeedbackSenderInterface* candidate_module,bool media_sender);// 取消当前活动的 REMB 模块,在需要切换或停止使用当前的 REMB 模块时,可以调用此方法void UnsetActiveRembModule();// 确定当前的活动 REMB 模块,可能会根据某些条件(如模块的状态、能力等)来选择合适的模块void DetermineActiveRembModule();// 将一个 RTP 发送模块添加到映射中,PacketRouter 可以管理多个 RTP 模块,并根据 SSRC 跟踪它们void AddSendRtpModuleToMap(RtpRtcpInterface* rtp_module, uint32_t ssrc);// 从映射中移除一个 RTP 发送模块void RemoveSendRtpModuleFromMap(uint32_t ssrc);// 确保某些操作(如对其他成员变量的访问)只在特定线程中执行,防止数据竞争和不一致的状态SequenceChecker thread_checker_;// Ssrc to RtpRtcpInterface module;// 哈希表,映射 SSRC(同步源标识符)到对应的 RTP 模块// 通过 SSRC,系统可以快速查找和访问与特定媒体流相关的 RTP 发送模块。这种映射支持高效的数据包路由和管理std::unordered_map<uint32_t, RtpRtcpInterface*> send_modules_map_RTC_GUARDED_BY(thread_checker_);// 存储指向 RTP 模块的指针// 用于维护添加的 RTP 模块的顺序,便于迭代和管理。这种数据结构在插入和删除操作上效率较高std::list<RtpRtcpInterface*> send_modules_list_RTC_GUARDED_BY(thread_checker_);// The last module used to send media.// 指向最后一个用于发送媒体的 RTP 模块的指针// 记录最近使用的 RTP 模块,可以在需要时快速访问。这对于优化发送过程和管理状态非常有用RtpRtcpInterface* last_send_module_ RTC_GUARDED_BY(thread_checker_);// Rtcp modules of the rtp receivers.// 存储指向 RTCP 反馈发送接口的指针// 这个向量可以包含多个 RTCP 反馈发送模块,用于处理反馈消息的发送,如带宽估计等。// 通过管理这些发送者,系统可以灵活地发送反馈信息std::vector<RtcpFeedbackSenderInterface*> rtcp_feedback_senders_RTC_GUARDED_BY(thread_checker_);// Candidates for the REMB module can be RTP sender/receiver modules, with// the sender modules taking precedence.// 发送REMB候选模块std::vector<RtcpFeedbackSenderInterface*> sender_remb_candidates_RTC_GUARDED_BY(thread_checker_);// 接收REMB候选模块std::vector<RtcpFeedbackSenderInterface*> receiver_remb_candidates_RTC_GUARDED_BY(thread_checker_);// 当前活动的REMB模块RtcpFeedbackSenderInterface* active_remb_module_RTC_GUARDED_BY(thread_checker_);// 用于跟踪传输序列uint64_t transport_seq_ RTC_GUARDED_BY(thread_checker_);// 用于在带宽估计变化时通知absl::AnyInvocable<void(RtpPacketToSend& packet,const PacedPacketInfo& pacing_info)>notify_bwe_callback_ RTC_GUARDED_BY(thread_checker_) = nullptr;// 待发送的前向纠错(FEC)数据包std::vector<std::unique_ptr<RtpPacketToSend>> pending_fec_packets_RTC_GUARDED_BY(thread_checker_);// 当前批处理中使用的 RTP 模块std::set<RtpRtcpInterface*> modules_used_in_current_batch_RTC_GUARDED_BY(thread_checker_);
};

PacketRouter中的核心函数SendPacket()的实现方式如下,其中调用了rtp_module->SendPacket()将RTP数据包送入RtpRtcpInterface中,rtp_module的数据类型为RtpRtcpInterface

void PacketRouter::SendPacket(std::unique_ptr<RtpPacketToSend> packet,const PacedPacketInfo& cluster_info) {// ...rtp_module->SendPacket(std::move(packet), cluster_info);modules_used_in_current_batch_.insert(rtp_module);// Sending succeeded.if (rtp_module->SupportsRtxPayloadPadding()) {// This is now the last module to send media, and has the desired// properties needed for payload based padding. Cache it for later use.last_send_module_ = rtp_module;}for (auto& packet : rtp_module->FetchFecPackets()) {pending_fec_packets_.push_back(std::move(packet));}
}

实际上,RtpRtcpInterface中并没有实现SendPacket()这个函数,而是声明为了纯虚函数,位于modules/rtp_rtcp/source/rtp_rtcp_interface.h

class RtpRtcpInterface : public RtcpFeedbackSenderInterface {public:// Try to send the provided packet. Returns true iff packet matches any of// the SSRCs for this module (media/rtx/fec etc) and was forwarded to the// transport.virtual bool TrySendPacket(std::unique_ptr<RtpPacketToSend> packet,const PacedPacketInfo& pacing_info) = 0;//  Returns true if the module can send media packets and the module is ready//  so send `packet` A RTP Sequence numbers may or may not have been assigned//  to the packet.virtual bool CanSendPacket(const RtpPacketToSend& packet) const = 0;//  Assigns continuous RTP sequence number to packet.virtual void AssignSequenceNumber(RtpPacketToSend& packet) = 0;// Send the packet to transport. Before using this method, a caller must// ensure the packet can be sent by first checking if the packet can be sent// using CanSendPacket and the packet must be assigned a sequence number using// AssignSequenceNumber.virtual void SendPacket(std::unique_ptr<RtpPacketToSend> packet,const PacedPacketInfo& pacing_info) = 0;}

RtpRtcpInterface中SendPacket()的实现由ModuleRtpRtcpImpl2给出,这个类以public的方式继承了RtpRtcpInterface

3.模块RtpRtcp的实现(ModuleRtpRtcpImpl2)

RtpRtcp模块负责处理 RTP数据包的发送和接收,而ModuleRtpRtcpImpl2是RtpRtcp模块的具体实现,其中实现了RtpRtcpInterface中最重要的函数SendPacket(),声明位于modules/rtp_rtcp/source/rtp_rtcp_impl2.h

class ModuleRtpRtcpImpl2 final : public RtpRtcpInterface,public RTCPReceiver::ModuleRtpRtcp {public:ModuleRtpRtcpImpl2(const Environment& env,const RtpRtcpInterface::Configuration& configuration);~ModuleRtpRtcpImpl2() override;// Receiver part.// Called when we receive an RTCP packet.// 接收到Rtcp数据包void IncomingRtcpPacket(rtc::ArrayView<const uint8_t> incoming_packet) override; // 设置远端SSRCvoid SetRemoteSSRC(uint32_t ssrc) override;// 设置本地SSRCvoid SetLocalSsrc(uint32_t local_ssrc) override;// Sender part.// 设置发送负载频率void RegisterSendPayloadFrequency(int payload_type,int payload_frequency) override;// 注销一个发送载荷类型int32_t DeRegisterSendPayload(int8_t payload_type) override;// 是否允许在RTP头部扩展中使用混合的extmap配置void SetExtmapAllowMixed(bool extmap_allow_mixed) override;// 注册一个RTP头部扩展void RegisterRtpHeaderExtension(absl::string_view uri, int id) override;// 注销一个RTP头部扩展void DeregisterSendRtpHeaderExtension(absl::string_view uri) override;bool SupportsPadding() const override;bool SupportsRtxPayloadPadding() const override;// Get start timestamp.// 获取开始时间戳uint32_t StartTimestamp() const override;// Configure start timestamp, default is a random number.// 设置开始时间戳void SetStartTimestamp(uint32_t timestamp) override;uint16_t SequenceNumber() const override;// Set SequenceNumber, default is a random number.void SetSequenceNumber(uint16_t seq) override;void SetRtpState(const RtpState& rtp_state) override;void SetRtxState(const RtpState& rtp_state) override;RtpState GetRtpState() const override;RtpState GetRtxState() const override;void SetNonSenderRttMeasurement(bool enabled) override;uint32_t SSRC() const override { return rtcp_sender_.SSRC(); }// Semantically identical to `SSRC()` but must be called on the packet// delivery thread/tq and returns the ssrc that maps to// RtpRtcpInterface::Configuration::local_media_ssrc.uint32_t local_media_ssrc() const;void SetMid(absl::string_view mid) override;RTCPSender::FeedbackState GetFeedbackState();void SetRtxSendStatus(int mode) override;int RtxSendStatus() const override;std::optional<uint32_t> RtxSsrc() const override;void SetRtxSendPayloadType(int payload_type,int associated_payload_type) override;std::optional<uint32_t> FlexfecSsrc() const override;// Sends kRtcpByeCode when going from true to false.// 设置发送状态int32_t SetSendingStatus(bool sending) override;// 检查当前是否在发送bool Sending() const override;// Drops or relays media packets.// 设置发送媒体状态void SetSendingMediaStatus(bool sending) override;// 检查是否处于发送媒体状态bool SendingMedia() const override;bool IsAudioConfigured() const override;void SetAsPartOfAllocation(bool part_of_allocation) override;// 发送RTP帧之前被调用。它提供了RTP帧的时间戳、捕获时间、载荷类型,以及一个标志来强制发送一个发送者报告bool OnSendingRtpFrame(uint32_t timestamp,int64_t capture_time_ms,int payload_type,bool force_sender_report) override;// 检查是否可以发送给定的RTP包。它根据当前的网络条件和发送策略来决定是否可以发送这个包bool CanSendPacket(const RtpPacketToSend& packet) const override;// 为RTP包分配一个序列号。RTP序列号是一个递增的整数,用于在接收端正确地排序和重组RTP流void AssignSequenceNumber(RtpPacketToSend& packet) override;// 发送数据包void SendPacket(std::unique_ptr<RtpPacketToSend> packet,const PacedPacketInfo& pacing_info) override;// 尝试发送数据包bool TrySendPacket(std::unique_ptr<RtpPacketToSend> packet,const PacedPacketInfo& pacing_info) override;// 在一批RTP包发送完成后被调用。它可以用来执行一些清理工作,或者更新发送器的状态void OnBatchComplete() override;// 设置前向纠错(FEC)保护参数void SetFecProtectionParams(const FecProtectionParams& delta_params,const FecProtectionParams& key_params) override;std::vector<std::unique_ptr<RtpPacketToSend>> FetchFecPackets() override;// 放弃重传操作void OnAbortedRetransmissions(rtc::ArrayView<const uint16_t> sequence_numbers) override;// RTP包被确认收void OnPacketsAcknowledged(rtc::ArrayView<const uint16_t> sequence_numbers) override;std::vector<std::unique_ptr<RtpPacketToSend>> GeneratePadding(size_t target_size_bytes) override;// 获取发送的RTP数据包信息std::vector<RtpSequenceNumberMap::Info> GetSentRtpPacketInfos(rtc::ArrayView<const uint16_t> sequence_numbers) const override;// ...// Force a send of an RTCP packet.// Normal SR and RR are triggered via the task queue that's current when this// object is created.// 发送RTCP数据包int32_t SendRTCP(RTCPPacketType rtcpPacketType) override;// 获取发送流的数据计数器void GetSendStreamDataCounters(StreamDataCounters* rtp_counters,StreamDataCounters* rtx_counters) const override;// A snapshot of the most recent Report Block with additional data of// interest to statistics. Used to implement RTCRemoteInboundRtpStreamStats.// Within this list, the `ReportBlockData::source_ssrc()`, which is the SSRC// of the corresponding outbound RTP stream, is unique.// 获取最新的接收报告块数据std::vector<ReportBlockData> GetLatestReportBlockData() const override;// 获取发送者报告统计信息std::optional<SenderReportStats> GetSenderReportStats() const override;// 获取非发送者往返时间(RTT)统计信息std::optional<NonSenderRttStats> GetNonSenderRttStats() const override;// (REMB) Receiver Estimated Max Bitrate.void SetRemb(int64_t bitrate_bps, std::vector<uint32_t> ssrcs) override;void UnsetRemb() override;void SetTmmbn(std::vector<rtcp::TmmbItem> bounding_set) override;// 获取Rtp最大Packet大小size_t MaxRtpPacketSize() const override;// 设置Rtp最大Packet大小void SetMaxRtpPacketSize(size_t max_packet_size) override;// (NACK) Negative acknowledgment part.// Send a Negative acknowledgment packet.// TODO(philipel): Deprecate SendNACK and use SendNack instead.int32_t SendNACK(const uint16_t* nack_list, uint16_t size) override;void SendNack(const std::vector<uint16_t>& sequence_numbers) override;// Store the sent packets, needed to answer to a negative acknowledgment// requests.// 设置是否存储已发送的RTP包,以及存储多少个包void SetStorePacketsStatus(bool enable, uint16_t number_to_store) override;// 发送一个组合的RTCP包void SendCombinedRtcpPacket(std::vector<std::unique_ptr<rtcp::RtcpPacket>> rtcp_packets) override;// Video part.// 发送丢包通知int32_t SendLossNotification(uint16_t last_decoded_seq_num,uint16_t last_received_seq_num,bool decodability_flag,bool buffering_allowed) override;// 获取发送速率RtpSendRates GetSendRates() const override;// 收到负确认(NACK)void OnReceivedNack(const std::vector<uint16_t>& nack_sequence_numbers) override;// 收到RTCP报告块void OnReceivedRtcpReportBlocks(rtc::ArrayView<const ReportBlockData> report_blocks) override;// 请求发送报告void OnRequestSendReport() override;// 设置视频比特率分配void SetVideoBitrateAllocation(const VideoBitrateAllocation& bitrate) override;RTPSender* RtpSender() override;const RTPSender* RtpSender() const override;private:FRIEND_TEST_ALL_PREFIXES(RtpRtcpImpl2Test, Rtt);FRIEND_TEST_ALL_PREFIXES(RtpRtcpImpl2Test, RttForReceiverOnly);// Rtp发送器上下文结构体struct RtpSenderContext {explicit RtpSenderContext(const Environment& env,TaskQueueBase& worker_queue,const RtpRtcpInterface::Configuration& config);// Storage of packets, for retransmissions and padding, if applicable.// packets的历史记录RtpPacketHistory packet_history;SequenceChecker sequencing_checker;// Handles sequence number assignment and padding timestamp generation.// 个负责RTP包序列号分配和填充时间戳生成的组件。它确保RTP包按照正确的顺序发送PacketSequencer sequencer RTC_GUARDED_BY(sequencing_checker);// Handles final time timestamping/stats/etc and handover to Transport.// 负责最终时间戳标记、统计信息收集等处理,并将RTP包交给传输层(Transport)的组件RtpSenderEgress packet_sender;// If no paced sender configured, this class will be used to pass packets// from `packet_generator_` to `packet_sender_`./*如果没有配置节流发送器(paced sender),则non_paced_sender将被用来将packet_generator_生成的包直接传递给packet_sender_。NonPacedPacketSender是一个不进行节流控制的发送器,它简单地将包从一个生成器传递到另一个发送器*/RtpSenderEgress::NonPacedPacketSender non_paced_sender;// Handles creation of RTP packets to be sent.// 负责创建要发送的RTP包的组件。它根据输入的媒体流(如音频或视频)生成RTP包,// 包括添加RTP头、有效载荷和其他必要的RTP信息RTPSender packet_generator;};void set_rtt_ms(int64_t rtt_ms);int64_t rtt_ms() const;bool TimeToSendFullNackList(int64_t now) const;// Called on a timer, once a second, on the worker_queue_, to update the RTT,// check if we need to send RTCP report, send TMMBR updates and fire events.// 周期性更新RTT,检查是否需要发送RTCP报告void PeriodicUpdate();// Returns true if the module is configured to store packets.// 当前的配置下是否会存储packetsbool StorePackets() const;// ...
};

这些函数和成员中,最重要的是结构体RtpSenderContext中的RtpSenderEgress packet_sender,这个成员调用了TrySendPacket()函数,将当前模块的packet送入到传输层进行传输。实现的方式如下,其中TrySendPacket()又调用了SendPacket()实现具体的发送任务

bool ModuleRtpRtcpImpl2::TrySendPacket(std::unique_ptr<RtpPacketToSend> packet,const PacedPacketInfo& pacing_info) {if (!packet || !CanSendPacket(*packet)) {return false;}AssignSequenceNumber(*packet);// 调用SendPacket()执行具体的发送任务SendPacket(std::move(packet), pacing_info);return true;
}void ModuleRtpRtcpImpl2::SendPacket(std::unique_ptr<RtpPacketToSend> packet,const PacedPacketInfo& pacing_info) {RTC_DCHECK_RUN_ON(&rtp_sender_->sequencing_checker);RTC_DCHECK(CanSendPacket(*packet));// 将packet送入到传输层,这里的rtp_sender_数据类型为RtpSenderEgressrtp_sender_->packet_sender.SendPacket(std::move(packet), pacing_info);
}

4.RTP出站发送器(RTPSenderEgress)

RTPSenderEgrees是RTP层面最后一个模块,其中的"Egress"翻译为”出站“或”出口“。这个模块的功能是将之前处理好的网络协议层信息(RTP和RTCP的信息)发送给传输层(Transport),交给其执行。这个类的声明位于modules/rtp_rtcp/source/rtp_sender_egress.h中

class RtpSenderEgress {public:// Helper class that redirects packets directly to the send part of this class// without passing through an actual paced sender.// 不进行节流的Packet发送器class NonPacedPacketSender : public RtpPacketSender {public:NonPacedPacketSender(TaskQueueBase& worker_queue,RtpSenderEgress* sender,PacketSequencer* sequencer);virtual ~NonPacedPacketSender();void EnqueuePackets(// ...};RtpSenderEgress(const Environment& env,const RtpRtcpInterface::Configuration& config,RtpPacketHistory* packet_history);~RtpSenderEgress();// 发送数据包到下一层级void SendPacket(std::unique_ptr<RtpPacketToSend> packet,const PacedPacketInfo& pacing_info);void OnBatchComplete();uint32_t Ssrc() const { return ssrc_; }std::optional<uint32_t> RtxSsrc() const { return rtx_ssrc_; }std::optional<uint32_t> FlexFecSsrc() const { return flexfec_ssrc_; }RtpSendRates GetSendRates(Timestamp now) const;void GetDataCounters(StreamDataCounters* rtp_stats,StreamDataCounters* rtx_stats) const;// 强制将发送数据包包含在比特率分配中void ForceIncludeSendPacketsInAllocation(bool part_of_allocation);// 媒体是否已被发送bool MediaHasBeenSent() const;// 设置媒体是否已被发送void SetMediaHasBeenSent(bool media_sent);// 设置时间戳偏移量void SetTimestampOffset(uint32_t timestamp);// For each sequence number in `sequence_number`, recall the last RTP packet// which bore it - its timestamp and whether it was the first and/or last// packet in that frame. If all of the given sequence numbers could be// recalled, return a vector with all of them (in corresponding order).// If any could not be recalled, return an empty vector.// 获取Rtp数据包信息std::vector<RtpSequenceNumberMap::Info> GetSentRtpPacketInfos(rtc::ArrayView<const uint16_t> sequence_numbers) const;void SetFecProtectionParameters(const FecProtectionParams& delta_params,const FecProtectionParams& key_params);std::vector<std::unique_ptr<RtpPacketToSend>> FetchFecPackets();// Clears pending status for these sequence numbers in the packet history.// 终止重传void OnAbortedRetransmissions(rtc::ArrayView<const uint16_t> sequence_numbers);private:// ...// Sends packet on to `transport_`, leaving the RTP module.// 将packet送入到transport模块,数据包离开RTP模块bool SendPacketToNetwork(const RtpPacketToSend& packet,const PacketOptions& options,const PacedPacketInfo& pacing_info);// 更新Rtp状态void UpdateRtpStats(Timestamp now,uint32_t packet_ssrc,RtpPacketMediaType packet_type,RtpPacketCounter counter,size_t packet_size);// ...
};

其中的核心函数为SendPacket()、CompleteSendPacket()和SendPacketToNetwork(),它们的调用关系是

SendPakcet()->CompleteSendPacket()->SendPacketToNetwork()

前面两个函数还会处理一些其他的信息,下面直接看最终出站的函数SendPacketToNetwork(),调用了transport_->SendRtp()将packet送入到transport模块中

bool RtpSenderEgress::SendPacketToNetwork(const RtpPacketToSend& packet,const PacketOptions& options,const PacedPacketInfo& pacing_info) {RTC_DCHECK_RUN_ON(worker_queue_);// SendRtp()将packet送入到transport模块中if (transport_ == nullptr || !transport_->SendRtp(packet, options)) {RTC_LOG(LS_WARNING) << "Transport failed to send packet.";return false;}env_.event_log().Log(std::make_unique<RtcEventRtpPacketOutgoing>(packet, pacing_info.probe_cluster_id));return true;
}

Transport的声明位于api/call/transport.h中,其中包括了发送Rtp和Rtcp两个数据包的纯虚函数

class Transport {public:virtual bool SendRtp(rtc::ArrayView<const uint8_t> packet,const PacketOptions& options) = 0;virtual bool SendRtcp(rtc::ArrayView<const uint8_t> packet) = 0;protected:virtual ~Transport() {}
};

这里的SendRtp()由MediaChannelUtil::TransportForMediaChannels实现

5.媒体通道的实现(MediaChannelUtil)

MediaChannelUtil当中提供了TransportForMediaChannels类,该类封装了媒体通道的发送和接收功能,使得媒体通道可以通过统一的接口与不同的传输层(如RTP/RTCP传输)进行交互。最核心的函数为SendRtp(),能够用于发送RTP数据包,

除了发送RTP包,MediaChannelUtil还提供了接收RTCP包的功能。这包括处理RTCP报告块、发送器报告等,以支持媒体通道的监控和质量控制。另外,可能涉及将媒体通道适配到不同的传输协议和配置中,以确保媒体数据能够正确、高效地传输。

MediaChannelUtil的声明位于media/base/media_channel_impl.h中

/
/ The `MediaChannelUtil` class provides functionality that is used by
// multiple MediaChannel-like objects, of both sending and receiving
// types.
class MediaChannelUtil {public:MediaChannelUtil(webrtc::TaskQueueBase* network_thread,bool enable_dscp = false);virtual ~MediaChannelUtil();// Returns the absolute sendtime extension id value from media channel.virtual int GetRtpSendTimeExtnId() const;// webrtc命名空间的transport类webrtc::Transport* transport() { return &transport_; }// Base methods to send packet using MediaChannelNetworkInterface.// These methods are used by some tests only.bool SendPacket(rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options);bool SendRtcp(rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options);int SetOption(MediaChannelNetworkInterface::SocketType type,rtc::Socket::Option opt,int option);// Functions that form part of one or more interface classes.// Not marked override, since this class does not inherit from the// interfaces.// Corresponds to the SDP attribute extmap-allow-mixed, see RFC8285.// Set to true if it's allowed to mix one- and two-byte RTP header extensions// in the same stream. The setter and getter must only be called from// worker_thread.void SetExtmapAllowMixed(bool extmap_allow_mixed);bool ExtmapAllowMixed() const;void SetInterface(MediaChannelNetworkInterface* iface);// Returns `true` if a non-null MediaChannelNetworkInterface pointer is held.// Must be called on the network thread.bool HasNetworkInterface() const;protected:bool DscpEnabled() const;void SetPreferredDscp(rtc::DiffServCodePoint new_dscp);private:// Implementation of the webrtc::Transport interface required// by Call().// webrtc::Transport的具体实现class TransportForMediaChannels : public webrtc::Transport {public:TransportForMediaChannels(webrtc::TaskQueueBase* network_thread,bool enable_dscp);virtual ~TransportForMediaChannels();// Implementation of webrtc::Transport// webrtc::Transport的具体实现,发送Rtp数据bool SendRtp(rtc::ArrayView<const uint8_t> packet,const webrtc::PacketOptions& options) override;// 发送Rtcp数据bool SendRtcp(rtc::ArrayView<const uint8_t> packet) override;// Not implementation of webrtc::Transportvoid SetInterface(MediaChannelNetworkInterface* iface);int SetOption(MediaChannelNetworkInterface::SocketType type,rtc::Socket::Option opt,int option);// 执行packet的发送bool DoSendPacket(rtc::CopyOnWriteBuffer* packet,bool rtcp,const rtc::PacketOptions& options);// 检查网络接口bool HasNetworkInterface() const {RTC_DCHECK_RUN_ON(network_thread_);return network_interface_ != nullptr;}// 查询DSCP是否被启用bool DscpEnabled() const { return enable_dscp_; }void SetPreferredDscp(rtc::DiffServCodePoint new_dscp);private:// This is the DSCP value used for both RTP and RTCP channels if DSCP is// enabled. It can be changed at any time via `SetPreferredDscp`.rtc::DiffServCodePoint PreferredDscp() const {RTC_DCHECK_RUN_ON(network_thread_);return preferred_dscp_;}// Apply the preferred DSCP setting to the underlying network interface RTP// and RTCP channels. If DSCP is disabled, then apply the default DSCP// value.void UpdateDscp() RTC_RUN_ON(network_thread_);int SetOptionLocked(MediaChannelNetworkInterface::SocketType type,rtc::Socket::Option opt,int option) RTC_RUN_ON(network_thread_);const rtc::scoped_refptr<webrtc::PendingTaskSafetyFlag> network_safety_RTC_PT_GUARDED_BY(network_thread_);webrtc::TaskQueueBase* const network_thread_;const bool enable_dscp_;MediaChannelNetworkInterface* network_interface_RTC_GUARDED_BY(network_thread_) = nullptr;rtc::DiffServCodePoint preferred_dscp_ RTC_GUARDED_BY(network_thread_) =rtc::DSCP_DEFAULT;};bool extmap_allow_mixed_ = false;TransportForMediaChannels transport_;
};

MediaChannelUtil::TransportForMediaChannels::SendRtp()的定义位于media/base/media_channel_impl.cc中,调用DoSendPacket()执行具体的packet发送任务

bool MediaChannelUtil::TransportForMediaChannels::SendRtp(rtc::ArrayView<const uint8_t> packet,const webrtc::PacketOptions& options) {auto send =[this, packet_id = options.packet_id,included_in_feedback = options.included_in_feedback,included_in_allocation = options.included_in_allocation,batchable = options.batchable,last_packet_in_batch = options.last_packet_in_batch,is_media = options.is_media,packet = rtc::CopyOnWriteBuffer(packet, kMaxRtpPacketLen)]() mutable {rtc::PacketOptions rtc_options;rtc_options.packet_id = packet_id;if (DscpEnabled()) {rtc_options.dscp = PreferredDscp();}rtc_options.info_signaled_after_sent.included_in_feedback =included_in_feedback;rtc_options.info_signaled_after_sent.included_in_allocation =included_in_allocation;rtc_options.info_signaled_after_sent.is_media = is_media;rtc_options.batchable = batchable;rtc_options.last_packet_in_batch = last_packet_in_batch;// 执行发送packetDoSendPacket(&packet, false, rtc_options);};// ...
}

DoSendPacket()的定义如下

bool MediaChannelUtil::TransportForMediaChannels::DoSendPacket(rtc::CopyOnWriteBuffer* packet,bool rtcp,const rtc::PacketOptions& options) {// ...// 如果不是Rtcp,则使用SendPacket()发送数据包return (!rtcp) ? network_interface_->SendPacket(packet, options): network_interface_->SendRtcp(packet, options);
}

这里的network_interface_的数据类型为MediaChannelNetworkInterface,这个类中只定义了一些纯虚函数,没有实现SendPacket(),SendPacket()的实现由BaseChannel给出

6.基础通道(BaseChannel)

BaseChannel是WebRTC中的通道层核心,为不同类型的媒体通道(如音频、视频和数据)提供了一个共同的接口和实现基础。另外,BaseChannel与PeerConnection和Transport层进行对接,管理底层的网络连接和数据传输。从RTP数据包传输的角度来说,BaseChannel中实现了将Transport模块中的packet送入到PeerConnection通道上。BaseChannel的声明位于pc/channel.h中,其中的pc表示”Peer Connection“。

由于BaseChannel的声明很长,这里截取部分

class BaseChannel : public ChannelInterface,// TODO(tommi): Consider implementing these interfaces// via composition.public MediaChannelNetworkInterface,public webrtc::RtpPacketSinkInterface {// ...// NetworkInterface implementation, called by MediaEngine// 发送packetbool SendPacket(rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options) override;// 发送Rtcp数据包bool SendRtcp(rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options) override;// From RtpTransportInternalvoid OnWritableState(bool writable);// 检测到网络路由案发生变化void OnNetworkRouteChanged(std::optional<rtc::NetworkRoute> network_route);// 带有选项的Packet数据包发送bool SendPacket(bool rtcp,rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options);
}

其中最重要的是SendPacket()函数

bool BaseChannel::SendPacket(bool rtcp,rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options) {// ...if (!srtp_active()) {if (srtp_required_) {// The audio/video engines may attempt to send RTCP packets as soon as the// streams are created, so don't treat this as an error for RTCP.// See: https://bugs.chromium.org/p/webrtc/issues/detail?id=6809// However, there shouldn't be any RTP packets sent before SRTP is set// up (and SetSend(true) is called).RTC_DCHECK(rtcp) << "Can't send outgoing RTP packet for " << ToString()<< " when SRTP is inactive and crypto is required";return false;}RTC_DLOG(LS_WARNING) << "Sending an " << (rtcp ? "RTCP" : "RTP")<< " packet without encryption for " << ToString()<< ".";}// ...// 如果不是rtcp,使用SendRtpPacket()传递packetreturn rtcp ? rtp_transport_->SendRtcpPacket(packet, options, PF_SRTP_BYPASS): rtp_transport_->SendRtpPacket(packet, options, PF_SRTP_BYPASS);
}

rtp_transport_的数据类型为RtpTransportInternal,这是一个基础类,其中只定义了SendRtpPacket()的纯虚函数

// TODO(zhihuang): Pass the `packet` by copy so that the original data
// wouldn't be modified.
virtual bool SendRtpPacket(rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options,int flags) = 0;virtual bool SendRtcpPacket(rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options,int flags) = 0;

从BaseChannel::SendPacket()中看,会先使用srtp_active(),考虑了Srtp的情况,所以这里SendRtpPacket()应该是由SrtpTransport实现

6.安全RTP传输(SrtpTransport)

SrtpTransport继承自RtpTransport,主要保障RTP的安全性,声明位于pc/srtp_transpory.h中

class SrtpTransport : public RtpTransport {public:SrtpTransport(bool rtcp_mux_enabled, const FieldTrialsView& field_trials);virtual ~SrtpTransport() = default;// 发送Rtp数据包bool SendRtpPacket(rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options,int flags) override;// 发送Rtcp数据包bool SendRtcpPacket(rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options,int flags) override;// The transport becomes active if the send_session_ and recv_session_ are// created.// 如果创建了发送和接收会话,则transport会被启用bool IsSrtpActive() const override;bool IsWritable(bool rtcp) const override;// Create new send/recv sessions and set the negotiated crypto keys for RTP// packet encryption. The keys can either come from SDES negotiation or DTLS// handshake.// 设置Rtp参数bool SetRtpParams(int send_crypto_suite,const rtc::ZeroOnFreeBuffer<uint8_t>& send_key,const std::vector<int>& send_extension_ids,int recv_crypto_suite,const rtc::ZeroOnFreeBuffer<uint8_t>& recv_key,const std::vector<int>& recv_extension_ids);// Create new send/recv sessions and set the negotiated crypto keys for RTCP// packet encryption. The keys can either come from SDES negotiation or DTLS// handshake.// 设置Rtcp参数bool SetRtcpParams(int send_crypto_suite,const rtc::ZeroOnFreeBuffer<uint8_t>& send_key,const std::vector<int>& send_extension_ids,int recv_crypto_suite,const rtc::ZeroOnFreeBuffer<uint8_t>& recv_key,const std::vector<int>& recv_extension_ids);// ...
}

SrtpTransport::SendRtpPacket()的定义位于pc/srtp_transport.cc中,其中最后会调用SendPacket()发送packet

bool SrtpTransport::SendRtpPacket(rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options,int flags) {// ...return SendPacket(/*rtcp=*/false, packet, updated_options, flags);
}

值得注意的是,SrtpTransport中并没有实现SendPacket()函数,而是继承自其父类RtpTransport中的SendPacket(),定义位于pc/rtp_transport.cc中

bool RtpTransport::SendPacket(bool rtcp,rtc::CopyOnWriteBuffer* packet,const rtc::PacketOptions& options,int flags) {rtc::PacketTransportInternal* transport = rtcp && !rtcp_mux_enabled_? rtcp_packet_transport_: rtp_packet_transport_;// 发送packetint ret = transport->SendPacket(packet->cdata<char>(), packet->size(),options, flags);if (ret != static_cast<int>(packet->size())) {if (set_ready_to_send_false_if_send_fail_) {// TODO: webrtc:361124449 - Remove SetReadyToSend if field trial// WebRTC-SetReadyToSendFalseIfSendFail succeed 2024-12-01.if (transport->GetError() == ENOTCONN) {RTC_LOG(LS_WARNING) << "Got ENOTCONN from transport.";SetReadyToSend(rtcp, false);}}return false;}return true;
}

在上面的调用之中,rtp_packet_transport_由RtpTransport::SetRtpPacketTransport()确定,这个函数会在DtlsSrtpTransport::SetDtlsTransports()中调用,所以这里的transport使用的SendPacket()函数由DtlsSrtpTransport这个类来实现

7.DTLS-SRTP 传输(DtlsSrtpTransport)

DtlsSrtpTransport结合了DTLS(Datagram Transport Layer Security)的密钥交换功能和 SRTP(Secure Real-time Transport Protocol)的媒体加密功能,为WebRTC提供安全保障。DTLS 用于在通信双方之间安全地协商加密密钥,而 SRTP 则使用这些密钥来加密实时媒体流(如音频和视频),一旦 DTLS 握手完成,DtlsSrtpTransport 确保通过该通道发送的 RTP 或 RTCP 数据被安全地加密和保护。这意味着数据在传输过程中不会被窃听或篡改,从而保护通信的隐私和完整性。这个类声明在pc/dtls_srtp_transport.h中

// The subclass of SrtpTransport is used for DTLS-SRTP. When the DTLS handshake
// is finished, it extracts the keying materials from DtlsTransport and
// configures the SrtpSessions in the base class.
/*SrtpTransport 的子类用于 DTLS-SRTP。当 DTLS 握手完成后,它从 DtlsTransport 中提取密钥材料,并在基类中配置 SrtpSessions
*/
class DtlsSrtpTransport : public SrtpTransport {public:DtlsSrtpTransport(bool rtcp_mux_enabled, const FieldTrialsView& field_trials);// Set P2P layer RTP/RTCP DtlsTransports. When using RTCP-muxing,// `rtcp_dtls_transport` is null.// 设置 P2P 层的 RTP/RTCP DtlsTransports。当使用 RTCP-muxing 时,`rtcp_dtls_transport` 为 nullvoid SetDtlsTransports(cricket::DtlsTransportInternal* rtp_dtls_transport,cricket::DtlsTransportInternal* rtcp_dtls_transport);// 启用或禁用RTCP多路复用void SetRtcpMuxEnabled(bool enable) override;// Set the header extension ids that should be encrypted.// 更新应该被加密的发送方RTP头部扩展ID列表void UpdateSendEncryptedHeaderExtensionIds(const std::vector<int>& send_extension_ids);// 更新应该被加密的接收方RTP头部扩展ID列表void UpdateRecvEncryptedHeaderExtensionIds(const std::vector<int>& recv_extension_ids);// 检测到DTLS状态发生变化void SetOnDtlsStateChange(std::function<void(void)> callback);// If `active_reset_srtp_params_` is set to be true, the SRTP parameters will// be reset whenever the DtlsTransports are reset.// 设置是否在DTLS传输被重置时主动重置SRTP参数void SetActiveResetSrtpParams(bool active_reset_srtp_params) {active_reset_srtp_params_ = active_reset_srtp_params;}private:// ...// Owned by the TransportController.cricket::DtlsTransportInternal* rtp_dtls_transport_ = nullptr;cricket::DtlsTransportInternal* rtcp_dtls_transport_ = nullptr;// The encrypted header extension IDs.std::optional<std::vector<int>> send_extension_ids_;std::optional<std::vector<int>> recv_extension_ids_;bool active_reset_srtp_params_ = false;std::function<void(void)> on_dtls_state_change_;
};

使用SetDtlsTransport()来配置RtpTransport::SendPacket()中的transport,如下所示

void DtlsSrtpTransport::SetDtlsTransports(cricket::DtlsTransportInternal* rtp_dtls_transport,cricket::DtlsTransportInternal* rtcp_dtls_transport) {// ...SetRtcpDtlsTransport(rtcp_dtls_transport);SetRtcpPacketTransport(rtcp_dtls_transport);RTC_LOG(LS_INFO) << "Setting RTP Transport on " << transport_name<< " transport " << rtp_dtls_transport;SetRtpDtlsTransport(rtp_dtls_transport);// 设置rtp_dtls_transport为RtpTransport::SendPacket()中的的transport// rtp_dtls_transport的数据类型为cricket::DtlsTransportInternal// SetRtpPacketTransport()的定义位于RtpTransport当中SetRtpPacketTransport(rtp_dtls_transport);MaybeSetupDtlsSrtp();
}

从代码中看,使用rtp_dtls_transport初始化,rtp_dtls_transport的数据类型为cricket::DtlsTransportInternal*,这是一个基础类,其实现由DtlsTransport给出。

在WebRTC中,存在两个DtlsTransport的定义,分别位于pc文件夹和p2p/base两个文件夹,这两者的作用有所区别,p2p/base下的dtls_transport.h提供了底层的DTLS传输功能,包括数据包的发送和接收,而pc下的 dtls_transport.h提供了与高层API交互的接口,用于管理和配置DTLS传输。这里需要使用p2p/base,执行RTP数据包的发送

8.DTLS传输(DtlsTransport)

DTLS传输定义位于p2p/base/dtls_transport.h中,声明了数据包发送和接收的函数

// This class provides a DTLS SSLStreamAdapter inside a TransportChannel-style
// packet-based interface, wrapping an existing TransportChannel instance
// (e.g a P2PTransportChannel)
/*这个类在TransportChannel风格的基于数据包的接口内提供了一个DTLS SSLStreamAdapter,包装了一个现有的TransportChannel实例(例如P2PTransportChannel)。
*/// Here's the way this works:
//
//   DtlsTransport {
//       SSLStreamAdapter* dtls_ {
//           StreamInterfaceChannel downward_ {
//               IceTransportInternal* ice_transport_;
//           }
//       }
//   }
//
//   - Data which comes into DtlsTransport from the underlying
//     ice_transport_ via OnReadPacket() is checked for whether it is DTLS
//     or not, and if it is, is passed to DtlsTransport::HandleDtlsPacket,
//     which pushes it into to downward_. dtls_ is listening for events on
//     downward_, so it immediately calls downward_->Read().
//
/*通过OnReadPacket()从底层ice_transport_传入DtlsTransport的数据会被检查是否为DTLS数据,如果是,会被传递给DtlsTransport::HandleDtlsPacket,该函数将其推送到downward_。dtls_监听downward_上的事件,因此它会立即调用downward_->Read()。
*///   - Data written to DtlsTransport is passed either to downward_ or directly
//     to ice_transport_, depending on whether DTLS is negotiated and whether
//     the flags include PF_SRTP_BYPASS
//
/*写入DtlsTransport的数据会被传递给downward_或者直接传递给ice_transport_,这取决于是否协商了DTLS以及标志是否包含PF_SRTP_BYPASS。*/
//   - The SSLStreamAdapter writes to downward_->Write() which translates it
//     into packet writes on ice_transport_.
//
/*SSLStreamAdapter 写入到 downward_->Write(),这将其转换为在 ice_transport_ 上的数据包写入
*/
// This class is not thread safe; all methods must be called on the same thread
// as the constructor.
// 这个类不是线程安全的;所有方法都必须在与构造函数相同的线程上调用
class DtlsTransport : public DtlsTransportInternal {public:// `ice_transport` is the ICE transport this DTLS transport is wrapping.  It// must outlive this DTLS transport.//// `crypto_options` are the options used for the DTLS handshake. This affects// whether GCM crypto suites are negotiated.//// `event_log` is an optional RtcEventLog for logging state changes. It should// outlive the DtlsTransport.DtlsTransport(IceTransportInternal* ice_transport,const webrtc::CryptoOptions& crypto_options,webrtc::RtcEventLog* event_log,rtc::SSLProtocolVersion max_version = rtc::SSL_PROTOCOL_DTLS_12);~DtlsTransport() override;DtlsTransport(const DtlsTransport&) = delete;DtlsTransport& operator=(const DtlsTransport&) = delete;// ...// Called to send a packet (via DTLS, if turned on).int SendPacket(const char* data,size_t size,const rtc::PacketOptions& options,int flags) override;// ...IceTransportInternal* ice_transport() override;// Underlying ice_transport, not owned by this class.IceTransportInternal* const ice_transport_;std::unique_ptr<rtc::SSLStreamAdapter> dtls_;  // The DTLS stream// ...
};

其中最核心的发送函数SendPacket()的定义如下

// Called from upper layers to send a media packet.
int DtlsTransport::SendPacket(const char* data,size_t size,const rtc::PacketOptions& options,int flags) {if (!dtls_active_) {// Not doing DTLS.// 不进行dtls,直接发送packetreturn ice_transport_->SendPacket(data, size, options);}switch (dtls_state()) {// ...case webrtc::DtlsTransportState::kConnected:if (flags & PF_SRTP_BYPASS) {RTC_DCHECK(!srtp_ciphers_.empty());if (!IsRtpPacket(rtc::MakeArrayView(reinterpret_cast<const uint8_t*>(data), size))) {return -1;}// ICE接口来发送packetreturn ice_transport_->SendPacket(data, size, options);} else {// ...}}
}

这里进一步使用ice_transport_来发送数据包,这是一个IceTransportInternal数据类型,其SendPacket()的实现由P2PTransportChannel给出

9.P2P传输通道(P2PTransportChannel)

// P2PTransportChannel manages the candidates and connection process to keep
// two P2P clients connected to each other.
class RTC_EXPORT P2PTransportChannel : public IceTransportInternal,public IceAgentInterface {public:static std::unique_ptr<P2PTransportChannel> Create(absl::string_view transport_name,int component,webrtc::IceTransportInit init);// For testing only.// TODO(zstein): Remove once AsyncDnsResolverFactory is required.P2PTransportChannel(absl::string_view transport_name,int component,PortAllocator* allocator,const webrtc::FieldTrialsView* field_trials = nullptr);~P2PTransportChannel() override;P2PTransportChannel(const P2PTransportChannel&) = delete;P2PTransportChannel& operator=(const P2PTransportChannel&) = delete;// ...// From TransportChannel:int SendPacket(const char* data,size_t len,const rtc::PacketOptions& options,int flags) override;// ...
}

SendPacket()的定义如下

/ Send data to the other side, using our selected connection.
int P2PTransportChannel::SendPacket(const char* data,size_t len,const rtc::PacketOptions& options,int flags) {// ...// 发送数据包,送入到网络中进行传输int sent = selected_connection_->Send(data, len, modified_options);// ...
}

selected_connection_->Send()中使用SendTo()发送数据包到给定的IP地址

int ProxyConnection::Send(const void* data,size_t size,const rtc::PacketOptions& options) {// ...int sent =port_->SendTo(data, size, remote_candidate_.address(), options, true);// ...
}

SendTo()函数根据具体的协议有所不同,例如UDP协议和TCP协议

int UDPPort::SendTo(const void* data,size_t size,const rtc::SocketAddress& addr,const rtc::PacketOptions& options,bool payload);int TCPPort::SendTo(const void* data,size_t size,const rtc::SocketAddress& addr,const rtc::PacketOptions& options,bool payload);

SendTo()之后就是具体的网络部分了,这样视频流发送过程中类的简单分析就结束了

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/diannao/60474.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

24/11/7 算法笔记 PCA主成分分析

假如我们的数据集是n维的&#xff0c;共有m个数据(x,x,...,x)。我们希望将这m个数据的维度从n维降到k维&#xff0c;希望这m个k维的数据集尽可能的代表原始数据集。我们知道数据从n维降到k维肯定会有损失&#xff0c;但是我们希望损失尽可能的小。那么如何让这k维的数据尽可能表…

JS 实现SSE通讯和了解SSE通讯

SSE 介绍&#xff1a; Server-Sent Events&#xff08;SSE&#xff09;是一种用于实现服务器向客户端实时推送数据的Web技术。与传统的轮询和长轮询相比&#xff0c;SSE提供了更高效和实时的数据推送机制。 SSE基于HTTP协议&#xff0c;允许服务器将数据以事件流&#xff08;…

C/C++每日一练:查找链表的中间节点

链表&#xff08;Linked List&#xff09; 链表是一种线性数据结构&#xff0c;由一系列节点&#xff08;Node&#xff09;通过指针链接在一起。与数组不同&#xff0c;链表中的元素在内存中不需要连续存储&#xff0c;每个节点包含两部分&#xff1a; 数据部分&#xff1a;存…

对称加密与非对称加密:密码学的基石及 RSA 算法详解

对称加密与非对称加密&#xff1a;密码学的基石及 RSA 算法详解 在当今数字化的时代&#xff0c;信息安全至关重要。对称加密和非对称加密作为密码学中的两种基本加密技术&#xff0c;为我们的数据安全提供了强大的保障。本文将深入探讨对称加密和非对称加密的特点、应用场景&…

PH47代码框架全局函数及功能类

PH47代码框架全局函数及功能类 概述 全局函数及功能类体系是PH47框架当中除了4个逻辑层之外最重要的组成部分之一&#xff0c;它们可以在 整个PH7 代码框架及用户代码中使用。常用全局函数及功能类为 PH7 代码框架提供了最常用和最基础的功能实现。 全局函数主要包含了对时间…

力扣 LeetCode 203. 移除链表元素(Day2:链表)

解题思路&#xff1a; 方法一&#xff1a;头节点和非头节点分开处理 方法二&#xff1a;定义一个dummy虚拟节点&#xff0c;后面的节点就可以采用相同的处理方式 注意&#xff1a; cur需要指向要删除的节点的上一个节点&#xff0c;因为要越过这一个被删除的节点 class Sol…

IEC60870-5-104 协议源码架构详细分析

IEC60870-5-104 协议源码架构 前言一、资源三、目录层级一二、目录层级二config/lib60870_config.hdependencies/READMEexamplesCMakeLists.txtcs101_master_balancedcs104_client_asyncmulti_client_servertls_clienttls_server说明 make这些文件的作用是否需要导入这些文件&a…

turbo c 2.0 画螺栓

代码; #include<graphics.h> void bolt(x0,y0,d,l) int x0,y0,d,l; {int x1,x2,x3,x4,x5,x6,x7,x8;int y1,y2,y3,y4,y5,r1,r2,b,c;if(l>2*d) b2*d;else b1;r11.5*d;r20.38*d;c0.1*d;x1x0-0.7*d;x2x0-0.61*d;x3x0-0.32*d;x4x00.8*d;x5x0l-b;x6x0l-c;x7x0l-0.05*d;x8x0…

网络服务综合项目-博客

一、运行环境&#xff1a; 主机主机名系统服务192.168.31.128Server-WebLinuxWeb192.168.31.129Server-NFS-DNSLinuxNFS 二、基础配置&#xff1a; 配置主机名开启防火墙并配置部分开启selinux并配置服务器之间使用ntp.aliyun.com进行时间同步服务器之间实现ssh免密登录 三…

leetcode86:分隔链表

给你一个链表的头节点 head 和一个特定值 x &#xff0c;请你对链表进行分隔&#xff0c;使得所有 小于 x 的节点都出现在 大于或等于 x 的节点之前。 你应当 保留 两个分区中每个节点的初始相对位置。 示例 1&#xff1a; 输入&#xff1a;head [1,4,3,2,5,2], x 3 输出&am…

Android Mobile Network Settings | APN 菜单加载异常

问题 从log看是有创建APN对应的Controller&#xff08;功能逻辑是ok的&#xff09;&#xff0c;但是Mobile Network Settings无法显示&#xff08;UI异常&#xff09;。 日志分析 看似APN 菜单已经创建了&#xff0c;实际上并没有显示。 11-12 07:01:28.150 8773 8773 D Pr…

上海市计算机学会竞赛平台2020年4月月赛丙组永恒的生命游戏

题目背景 2020年4月11日&#xff0c;英国数学家 约翰霍顿康威&#xff08;John Horton Conway&#xff09;因为新型冠状病毒肺炎不幸逝世。他在群论、数论、代数、几何拓扑、理论物理、组合博弈论和几何等领域&#xff0c;都做出了重大贡献。他的离去是人类文明的损失。他最著…

FS8x 功能安全

fail-safe是电独立的和物理隔离的。fail-safe由自己的参考电压和电流提供,有自己的振荡器,有重复的模拟路径以最小化常见的故障,并有LBIST/ABIST来覆盖潜在故障。fail-safe根据设备部件号提供ASIL B或ASIL D遵从性。除非另有规定,fail-safe定时来自故障安全振荡器,其精度为…

项目模块十七:HttpServer模块

一、项目模块设计思路 目的&#xff1a;实现HTTP服务器搭建 思想&#xff1a;设计请求路由表&#xff0c;记录请求方法与对应业务的处理函数映射关系。用户实现请求方法和处理函数添加到路由表&#xff0c;服务器只接受请求并调用用户的处理函数即可。 处理流程&#xff1a; …

内网域环境、工作组、局域网等探针方案

1. 信息收集 1.1 网络收集 了解当前服务器的计算机基本信息&#xff0c;为后续判断服务器角色&#xff0c;网络环境做准备 systeminfo 详细信息 net start 启动服务 tasklist 进程列表 schtasks 计划任务&#xff08;受权限影响&#xff09; 了解当前服务器的网络接口信息…

什么是量化交易

课程大纲 内容初级初识量化&#xff0c;理解量化 初识量化 传统量化和AI量化的区别 量化思想挖掘 量化思想的挖掘及积累技巧 量化代码基础&#xff1a; python、pandas、SQL基础语法 金融数据分析 常用金融分析方式 常用因子分析方式 数据分析实战练习 回测及交易引擎 交易引擎…

OpenHarmony-1.启动流程

OpenHarmony启动流程 1.kernel的启动 流程图如下所示&#xff1a;   OpenHarmony(简称OH)的标准系统的底层系统是linux&#xff0c;所以调用如下代码&#xff1a; linux-5.10/init/main.c: noinline void __ref rest_init(void) {struct task_struct *tsk;int pid;rcu_sch…

【LeetCode】【算法】64. 最小路径和

LeetCode 64. 最小路径和 题目描述 给定一个包含非负整数的 m x n 网格 grid &#xff0c;请找出一条从左上角到右下角的路径&#xff0c;使得路径上的数字总和为最小。 说明&#xff1a;每次只能向下或者向右移动一步。 思路 思路&#xff1a;这种题太典了&#xff0c;典…

1.7 JS性能优化

从输入url到页面加载完成都做了些什么 输入 URL - 资源定位符 http://www.zhaowa.com - http 协议 追问&#xff1a;http 与 TCP 1. http - 应用层 < > TCP - 传输层 2. 关联 - http基于TCP实现连接 < > UDP > 握手 & 挥手 &#xff08;传输速率上较…

Spring Task详细讲解

✨Spring Task简介 Spring Task 是 Spring 提供的轻量级定时任务工具&#xff0c;也就意味着不需要再添加第三方依赖了&#xff0c;相比其他第三方类库更加方便易用。可以按照约定的时间自动执行某个代码逻辑。 使用场景&#xff1a; 信用卡每月还款提醒银行贷款每月还款提醒…